ποΈGitΠ―ΡΠ°ποΈ
Commit 35fba4d56d1a6fa0a2f44fee9c9a550c476b8c96
Parents : 1230c36
Author : Jeremiah K <17190268+jeremiah-k@users.noreply.github.com>
Signature : Signature validation error
Date : 2026-06-16T17:26:00-05:00
Committer : GitHub <noreply@github.com>
Date : 2026-06-16T17:26:00-05:00
fix(ble): Harden BLE connection lifecycle (#5795)
Co-authored-by: James Rich <james.a.rich@gmail.com>
Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Co-authored-by: James Rich <2199651+jamesarich@users.noreply.github.com>
Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Signed-off-by: James Rich <james.a.rich@gmail.com>
Changes
14 files changed, 2373 insertions(+), 150 deletions(-)
Diff
diff --git a/core/ble/src/commonMain/kotlin/org/meshtastic/core/ble/BleExceptionClassifier.kt b/core/ble/src/commonMain/kotlin/org/meshtastic/core/ble/BleExceptionClassifier.kt
index 2c5c107a9d..54f5509216 100644
--- a/core/ble/src/commonMain/kotlin/org/meshtastic/core/ble/BleExceptionClassifier.kt
+++ b/core/ble/src/commonMain/kotlin/org/meshtastic/core/ble/BleExceptionClassifier.kt
@@ -62,3 +62,47 @@ fun Throwable.classifyBleException(): BleExceptionInfo? = when (this) {
else -> null
}
+
+/**
+ * GATT status codes that indicate the BLE session is irrecoverably broken.
+ *
+ * Used by [isSessionFatalBleException] and shared across the BLE stack so classification stays in one place.
+ */
+@Suppress("MagicNumber")
+private val FATAL_GATT_STATUSES =
+ setOf(
+ // 0x08 β link-layer supervision timeout (peer out of range or asleep)
+ 8, // GATT_CONN_TIMEOUT
+ // 0x13 β peer-initiated disconnect (firmware reboot, user power-cycle, nRF52 link drop).
+ // The most common Meshtastic disconnect signal; without it fromRadio would spin-retry a dead link.
+ 19, // GATT_CONN_TERMINATE_PEER_USER
+ // 0x16 β link manager protocol timeout (radio firmware/hardware hang)
+ 22, // GATT_CONN_LMP_TIMEOUT
+ // 0x3E β connection establishment failed (discovered during connect handshake)
+ 62, // GATT_CONN_FAIL_ESTABLISH
+ // 0x85 β generic connection failure; commonly fires at runtime against a stale GATT handle
+ 133, // GATT_ERROR
+ // 0x81 β unrecoverable operation failure
+ 129, // GATT_FAILURE
+ )
+
+/**
+ * Returns `true` if this throwable indicates the BLE session is irrecoverably broken and should be torn down
+ * (triggering reconnection), as opposed to a transient condition that can be retried.
+ *
+ * Also checks the cause chain β if a session-fatal exception is wrapped inside another exception (e.g., by coroutine
+ * machinery or retry logic), it is still detected. Depth-limited to prevent stack overflow on malformed cause chains.
+ */
+fun Throwable.isSessionFatalBleException(): Boolean = isSessionFatalBleExceptionInternal(maxDepth = 10)
+
+private fun Throwable.isSessionFatalBleExceptionInternal(maxDepth: Int): Boolean {
+ if (maxDepth <= 0) return false
+ return when (this) {
+ is NotConnectedException -> true
+
+ is GattStatusException ->
+ status in FATAL_GATT_STATUSES || cause?.isSessionFatalBleExceptionInternal(maxDepth - 1) ?: false
+
+ else -> cause?.isSessionFatalBleExceptionInternal(maxDepth - 1) ?: false
+ }
+}
diff --git a/core/ble/src/commonMain/kotlin/org/meshtastic/core/ble/KableMeshtasticRadioProfile.kt b/core/ble/src/commonMain/kotlin/org/meshtastic/core/ble/KableMeshtasticRadioProfile.kt
index 8ecb253bf2..8365f8fe79 100644
--- a/core/ble/src/commonMain/kotlin/org/meshtastic/core/ble/KableMeshtasticRadioProfile.kt
+++ b/core/ble/src/commonMain/kotlin/org/meshtastic/core/ble/KableMeshtasticRadioProfile.kt
@@ -25,7 +25,8 @@ import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.flow.MutableSharedFlow
import kotlinx.coroutines.flow.catch
import kotlinx.coroutines.flow.channelFlow
-import kotlinx.coroutines.flow.emptyFlow
+import kotlinx.coroutines.flow.emitAll
+import kotlinx.coroutines.flow.flow
import kotlinx.coroutines.launch
import org.meshtastic.core.ble.MeshtasticBleConstants.FROMNUM_CHARACTERISTIC
import org.meshtastic.core.ble.MeshtasticBleConstants.FROMRADIO_CHARACTERISTIC
@@ -70,12 +71,23 @@ class KableMeshtasticRadioProfile(private val service: BleService) : MeshtasticR
override val fromRadio: Flow<ByteArray> = channelFlow {
launch {
if (service.hasCharacteristic(fromNum)) {
- service
- .observe(fromNum) {
- Logger.d { "FROMNUM CCCD written β notifications enabled" }
- subscriptionReady.complete(Unit)
- }
- .collect { triggerDrain.tryEmit(Unit) }
+ try {
+ service
+ .observe(fromNum) {
+ Logger.d { "FROMNUM CCCD written β notifications enabled" }
+ subscriptionReady.complete(Unit)
+ }
+ .collect { triggerDrain.tryEmit(Unit) }
+ } catch (e: CancellationException) {
+ // Propagate cancellation β don't complete subscription as that would
+ // let setup proceed on a cancelled scope.
+ throw e
+ } catch (e: Exception) {
+ // Complete subscriptionReady exceptionally so awaitSubscriptionReady()
+ // throws promptly instead of waiting for the transport's 5s timeout.
+ subscriptionReady.completeExceptionally(e)
+ throw e
+ }
} else {
subscriptionReady.complete(Unit)
}
@@ -94,7 +106,13 @@ class KableMeshtasticRadioProfile(private val service: BleService) : MeshtasticR
} catch (e: CancellationException) {
throw e
} catch (e: Exception) {
- Logger.w(e) { "FROMRADIO read error, pausing before next drain trigger" }
+ // Session-fatal BLE exceptions must propagate so the transport layer detects the
+ // broken session via its .catch handler and triggers reconnection.
+ if (e.isSessionFatalBleException()) {
+ Logger.w(e) { "FROMRADIO read hit session-fatal BLE exception β propagating for reconnect" }
+ throw e
+ }
+ Logger.w(e) { "FROMRADIO read error (transient), pausing before next drain trigger" }
keepReading = false
delay(TRANSIENT_RETRY_DELAY)
}
@@ -102,16 +120,27 @@ class KableMeshtasticRadioProfile(private val service: BleService) : MeshtasticR
}
}
- override val logRadio: Flow<ByteArray> =
- if (service.hasCharacteristic(logRadioChar)) {
+ /**
+ * Observes the LOGRADIO characteristic. Session-fatal exceptions propagate to trigger reconnect.
+ *
+ * Note: a transient (non-fatal) observation error terminates this flow permanently for the current session. Unlike
+ * [fromRadio] which has a retry loop, [logRadio] does not recover from transient errors. This is intentional β
+ * logRadio is diagnostic-only, and the flow is recreated on the next reconnect cycle.
+ */
+ override val logRadio: Flow<ByteArray> = flow {
+ if (!service.hasCharacteristic(logRadioChar)) return@flow
+ emitAll(
service.observe(logRadioChar).catch { e ->
if (e is CancellationException) throw e
+ if (e.isSessionFatalBleException()) {
+ Logger.w(e) { "logRadio observation hit session-fatal BLE exception β propagating for reconnect" }
+ throw e
+ }
// logRadio is optional β log at debug for diagnostics but don't surface to callers.
Logger.d(e) { "logRadio observation failure suppressed" }
- }
- } else {
- emptyFlow()
- }
+ },
+ )
+ }
override suspend fun sendToRadio(packet: ByteArray) {
service.write(toRadio, packet, toRadioWriteType)
diff --git a/core/ble/src/commonTest/kotlin/org/meshtastic/core/ble/BleExceptionClassifierTest.kt b/core/ble/src/commonTest/kotlin/org/meshtastic/core/ble/BleExceptionClassifierTest.kt
index 1170b973b7..521cf9d78b 100644
--- a/core/ble/src/commonTest/kotlin/org/meshtastic/core/ble/BleExceptionClassifierTest.kt
+++ b/core/ble/src/commonTest/kotlin/org/meshtastic/core/ble/BleExceptionClassifierTest.kt
@@ -64,4 +64,57 @@ class BleExceptionClassifierTest {
fun `RuntimeException returns null`() {
assertNull(RuntimeException("boom").classifyBleException())
}
+
+ // --- isSessionFatalBleException tests ---
+
+ @Test
+ fun `isSessionFatalBleException returns true for NotConnectedException`() {
+ assertTrue(NotConnectedException("test").isSessionFatalBleException())
+ }
+
+ @Test
+ fun `isSessionFatalBleException returns true for fatal GATT status codes`() {
+ assertTrue(GattStatusException(status = 133, message = "test").isSessionFatalBleException())
+ assertTrue(GattStatusException(status = 8, message = "test").isSessionFatalBleException())
+ assertTrue(GattStatusException(status = 129, message = "test").isSessionFatalBleException())
+ }
+
+ @Test
+ fun `isSessionFatalBleException returns true for peer-disconnect and establishment-failure codes`() {
+ // 19 = GATT_CONN_TERMINATE_PEER_USER β firmware reboot / peer-initiated disconnect
+ assertTrue(GattStatusException(status = 19, message = "peer disconnect").isSessionFatalBleException())
+ // 22 = GATT_CONN_LMP_TIMEOUT β link manager protocol timeout
+ assertTrue(GattStatusException(status = 22, message = "lmp timeout").isSessionFatalBleException())
+ // 62 = GATT_CONN_FAIL_ESTABLISH β connection establishment failed
+ assertTrue(GattStatusException(status = 62, message = "establish failed").isSessionFatalBleException())
+ }
+
+ @Test
+ fun `isSessionFatalBleException returns false for transient GATT status`() {
+ assertFalse(GattStatusException(status = 6, message = "busy").isSessionFatalBleException())
+ }
+
+ @Test
+ fun `isSessionFatalBleException returns false for unrelated exceptions`() {
+ assertFalse(IllegalStateException("test").isSessionFatalBleException())
+ assertFalse(RuntimeException("test").isSessionFatalBleException())
+ }
+
+ @Test
+ fun `isSessionFatalBleException traverses cause chain for wrapped exceptions`() {
+ val fatal = GattStatusException(status = 133, message = "wrapped fatal")
+ val wrapper = RuntimeException("wrapper", fatal)
+ assertTrue(wrapper.isSessionFatalBleException())
+
+ val notConnected = NotConnectedException("wrapped")
+ val doubleWrapper = IllegalStateException("outer", RuntimeException("middle", notConnected))
+ assertTrue(doubleWrapper.isSessionFatalBleException())
+ }
+
+ @Test
+ fun `isSessionFatalBleException returns false for non-fatal cause chain`() {
+ val transient = GattStatusException(status = 6, message = "busy")
+ val wrapper = RuntimeException("wrapper", transient)
+ assertFalse(wrapper.isSessionFatalBleException())
+ }
}
diff --git a/core/ble/src/commonTest/kotlin/org/meshtastic/core/ble/KableMeshtasticRadioProfileExceptionTest.kt b/core/ble/src/commonTest/kotlin/org/meshtastic/core/ble/KableMeshtasticRadioProfileExceptionTest.kt
new file mode 100644
index 0000000000..7afd273856
--- /dev/null
+++ b/core/ble/src/commonTest/kotlin/org/meshtastic/core/ble/KableMeshtasticRadioProfileExceptionTest.kt
@@ -0,0 +1,161 @@
+/*
+ * Copyright (c) 2026 Meshtastic LLC
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program. If not, see <https://www.gnu.org/licenses/>.
+ */
+package org.meshtastic.core.ble
+
+import com.juul.kable.GattStatusException
+import com.juul.kable.NotConnectedException
+import kotlinx.coroutines.CancellationException
+import kotlinx.coroutines.ExperimentalCoroutinesApi
+import kotlinx.coroutines.flow.collect
+import kotlinx.coroutines.flow.first
+import kotlinx.coroutines.launch
+import kotlinx.coroutines.test.advanceTimeBy
+import kotlinx.coroutines.test.advanceUntilIdle
+import kotlinx.coroutines.test.runTest
+import org.meshtastic.core.testing.FakeBleService
+import kotlin.test.Test
+import kotlin.test.assertFailsWith
+import kotlin.test.assertTrue
+
+/**
+ * Tests that [KableMeshtasticRadioProfile.fromRadio] propagates session-fatal BLE exceptions (so the transport layer
+ * can detect broken sessions) while suppressing transient errors.
+ */
+@OptIn(ExperimentalCoroutinesApi::class)
+class KableMeshtasticRadioProfileExceptionTest {
+
+ private fun createService(): FakeBleService = FakeBleService().apply {
+ addCharacteristic(MeshtasticBleConstants.FROMNUM_CHARACTERISTIC)
+ addCharacteristic(MeshtasticBleConstants.FROMRADIO_CHARACTERISTIC)
+ addCharacteristic(MeshtasticBleConstants.TORADIO_CHARACTERISTIC)
+ }
+
+ @Test
+ fun `fromRadio propagates NotConnectedException to collector`() = runTest {
+ val service = createService()
+ val profile = KableMeshtasticRadioProfile(service)
+
+ // Set up the read to throw NotConnectedException immediately
+ service.readException = NotConnectedException("session closed")
+
+ val result = assertFailsWith<NotConnectedException> { profile.fromRadio.first() }
+ assertTrue(result.message!!.contains("session closed"))
+ }
+
+ @Test
+ fun `fromRadio propagates fatal GattStatusException to collector`() = runTest {
+ val service = createService()
+ val profile = KableMeshtasticRadioProfile(service)
+
+ // Fatal GATT status (133 = GATT_ERROR, in FATAL_GATT_STATUSES)
+ service.readException = GattStatusException(message = "GATT error", status = 133)
+
+ val result = assertFailsWith<GattStatusException> { profile.fromRadio.first() }
+ assertTrue(result.message!!.contains("GATT error"))
+ }
+
+ @Test
+ fun `fromRadio propagates CancellationException`() = runTest {
+ val service = createService()
+ val profile = KableMeshtasticRadioProfile(service)
+
+ // Set up the read to throw CancellationException
+ service.readException = CancellationException("cancelled")
+
+ val result = assertFailsWith<CancellationException> { profile.fromRadio.first() }
+ assertTrue(result.message!!.contains("cancelled"))
+ }
+
+ @Test
+ fun `fromRadio suppresses transient GattStatusException and continues`() = runTest {
+ val service = createService()
+ val profile = KableMeshtasticRadioProfile(service)
+
+ // Non-fatal GATT status (e.g. status 6 = GATT_BUSY, not in FATAL_GATT_STATUSES)
+ service.readException = GattStatusException(message = "transient busy", status = 6)
+
+ // Collect from the flow in a coroutine; after the transient error is suppressed and the
+ // delay elapses, the next drain trigger should restart reading. Enqueue a real packet so
+ // the collector eventually receives something.
+ var collected = false
+ val collectJob = launch { profile.fromRadio.collect { collected = true } }
+ advanceUntilIdle()
+
+ // The transient error was suppressed (500ms delay). Advance past it.
+ advanceTimeBy(600)
+
+ // Now enqueue a packet so the next drain cycle emits data, proving the flow survived.
+ service.enqueueRead(MeshtasticBleConstants.FROMRADIO_CHARACTERISTIC, byteArrayOf(42))
+ service.enqueueRead(MeshtasticBleConstants.FROMRADIO_CHARACTERISTIC, ByteArray(0))
+ profile.requestDrain()
+ advanceUntilIdle()
+
+ assertTrue(collected, "Flow should have emitted a packet after transient error was suppressed")
+
+ collectJob.cancel()
+ }
+
+ // --- logRadio fatal exception propagation tests ---
+
+ @Test
+ fun `logRadio propagates NotConnectedException from observation`() = runTest {
+ val service = createService().apply { addCharacteristic(MeshtasticBleConstants.LOGRADIO_CHARACTERISTIC) }
+ val profile = KableMeshtasticRadioProfile(service)
+ service.observeException = NotConnectedException("log radio session closed")
+ assertFailsWith<NotConnectedException> { profile.logRadio.first() }
+ }
+
+ @Test
+ fun `logRadio propagates fatal GattStatusException from observation`() = runTest {
+ val service = createService().apply { addCharacteristic(MeshtasticBleConstants.LOGRADIO_CHARACTERISTIC) }
+ val profile = KableMeshtasticRadioProfile(service)
+ service.observeException = GattStatusException(status = 133, message = "GATT error")
+ assertFailsWith<GattStatusException> { profile.logRadio.first() }
+ }
+
+ // --- subscriptionReady exceptional completion tests ---
+
+ @Test
+ fun `awaitSubscriptionReady throws promptly when FROMNUM observe fails before readiness`() = runTest {
+ val service = createService()
+ val profile = KableMeshtasticRadioProfile(service)
+
+ // Set observeException β when fromRadio is collected, the FROMNUM observe will throw
+ // before subscriptionReady is completed. The fix completes it exceptionally.
+ service.observeException = NotConnectedException("observe failed before CCCD")
+
+ // Start collecting fromRadio in a regular child coroutine so it's properly scoped
+ // and cancelled by the test framework. The exception is caught inside the coroutine,
+ // so it won't crash the test scope.
+ val collectJob = launch {
+ try {
+ profile.fromRadio.collect {}
+ } catch (e: Exception) {
+ // Expected β the observe failure propagates through the channelFlow
+ }
+ }
+ try {
+ advanceUntilIdle()
+
+ // awaitSubscriptionReady should throw the exception promptly, not hang
+ val result = assertFailsWith<NotConnectedException> { profile.awaitSubscriptionReady() }
+ assertTrue(result.message!!.contains("observe failed before CCCD"))
+ } finally {
+ collectJob.cancel()
+ }
+ }
+}
diff --git a/core/network/build.gradle.kts b/core/network/build.gradle.kts
index b9fbc1589e..7ee3655970 100644
--- a/core/network/build.gradle.kts
+++ b/core/network/build.gradle.kts
@@ -37,6 +37,7 @@ kotlin {
implementation(libs.okio)
api(libs.meshtastic.mqtt.client)
implementation(libs.kotlinx.serialization.json)
+ implementation(libs.kotlinx.atomicfu)
implementation(libs.ktor.client.core)
implementation(libs.ktor.client.content.negotiation)
implementation(libs.ktor.client.logging)
@@ -58,6 +59,7 @@ kotlin {
commonTest.dependencies {
implementation(projects.core.testing)
implementation(libs.kotlinx.coroutines.test)
+ implementation(libs.kable.core) // Kable exception types for BLE failure-injection tests
}
}
}
diff --git a/core/network/src/commonMain/kotlin/org/meshtastic/core/network/radio/BleRadioTransport.kt b/core/network/src/commonMain/kotlin/org/meshtastic/core/network/radio/BleRadioTransport.kt
index 37b90d526e..8acbefe9ab 100644
--- a/core/network/src/commonMain/kotlin/org/meshtastic/core/network/radio/BleRadioTransport.kt
+++ b/core/network/src/commonMain/kotlin/org/meshtastic/core/network/radio/BleRadioTransport.kt
@@ -19,6 +19,7 @@
package org.meshtastic.core.network.radio
import co.touchlab.kermit.Logger
+import kotlinx.atomicfu.atomic
import kotlinx.coroutines.CancellationException
import kotlinx.coroutines.CoroutineExceptionHandler
import kotlinx.coroutines.CoroutineScope
@@ -75,9 +76,27 @@ private val CONNECTION_TIMEOUT = 15.seconds
*/
private val HEARTBEAT_DRAIN_DELAY = 200.milliseconds
+private val TARGETED_SCAN_TIMEOUT = 2.seconds
private val SCAN_TIMEOUT = 5.seconds
private val GATT_CLEANUP_TIMEOUT = 5.seconds
+/**
+ * Bounded wait for the connectionState StateFlow to reflect Connected after connectAndAwait returns.
+ *
+ * In normal operation the observer coroutine lags by milliseconds. If a fatal session failure fires before Connected is
+ * observed and the StateFlow was already at a stale Disconnected value (no re-emit), this timeout prevents
+ * [attemptConnection] from hanging indefinitely.
+ */
+private val CONNECTED_GATE_TIMEOUT = 5.seconds
+
+/**
+ * Bounded wait for FROMNUM CCCD subscription before proceeding with the handshake.
+ *
+ * In normal operation the observe callback fires within milliseconds. If a fatal fromRadio failure fires before
+ * subscriptionReady completes, this timeout prevents a permanent hang.
+ */
+private val SUBSCRIPTION_READY_TIMEOUT = 5.seconds
+
/**
* Delay after onConnect before downgrading BLE connection priority to Balanced.
*
@@ -123,6 +142,18 @@ class BleRadioTransport(
private val exceptionHandler = CoroutineExceptionHandler { _, throwable ->
Logger.w(throwable) { "[$address] Uncaught exception in connectionScope" }
+ if (throwable !is CancellationException) {
+ // Record the cause BEFORE the CAS so there is no window where another coroutine
+ // sees sessionFailed == true but sessionFailureCause == null. first-cause is
+ // preserved: a concurrent loser only overwrites if still null.
+ recordSessionFailureCause(throwable)
+ if (sessionFailed.compareAndSet(expect = false, update = true)) {
+ radioService = null
+ isFullyConnected = false
+ val (isPermanent, msg) = throwable.toDisconnectReason()
+ callback.onDisconnect(isPermanent, errorMessage = msg)
+ }
+ }
cleanupScope.launch {
try {
bleConnection.disconnect()
@@ -130,8 +161,6 @@ class BleRadioTransport(
Logger.w(e) { "[$address] Failed to disconnect in exception handler" }
}
}
- val (isPermanent, msg) = throwable.toDisconnectReason()
- callback.onDisconnect(isPermanent, errorMessage = msg)
}
private val connectionScope: CoroutineScope =
@@ -141,17 +170,30 @@ class BleRadioTransport(
@Volatile private var connectionStartTime: Long = 0
- @Volatile private var packetsReceived: Int = 0
+ private val packetsReceived = atomic(0)
- @Volatile private var packetsSent: Int = 0
+ private val packetsSent = atomic(0)
- @Volatile private var bytesReceived: Long = 0
+ private val bytesReceived = atomic(0L)
- @Volatile private var bytesSent: Long = 0
+ private val bytesSent = atomic(0L)
@Volatile private var isFullyConnected = false
private var connectionJob: Job? = null
+ // Guards against duplicate callbacks when multiple writes or a write + fromRadio failure
+ // fire against the same stale session. Reset at the start of each attemptConnection().
+ // Write-path failures are serialized by writeMutex; fromRadio/logRadio .catch handlers run
+ // on the flow collector coroutine and may race. Atomic CAS guarantees first-writer-wins β
+ // only the caller that wins the compareAndSet(false, true) fires onDisconnect.
+ private val sessionFailed = atomic(false)
+
+ // Captures the exception that caused handleFailure() to tear down the session, so
+ // attemptConnection() can distinguish an internal-failure disconnect from a genuine
+ // LocalDisconnect (user-initiated or clean close). When non-null, the reconnect policy
+ // treats the disconnect as unintentional and applies backoff escalation.
+ @Volatile private var sessionFailureCause: Throwable? = null
+
// Never give up while the user has this device selected. Higher layers (SharedRadioInterfaceService)
// own the explicit-disconnect lifecycle and will close() us when the user picks a different device or
// toggles the connection off; until then, retry forever with the policy's exponential-backoff cap (60 s).
@@ -173,13 +215,21 @@ class BleRadioTransport(
// --- Connection & Discovery Logic ---
- /** Robustly finds the device. First checks bonded devices, then performs a short scan if not found. */
+ /** Robustly finds the device. Checks bonded devices, preferring a fresh scan result when available. */
+ @Suppress("ReturnCount")
private suspend fun findDevice(): BleDevice {
- bluetoothRepository.state.value.bondedDevices
- .firstOrNull { it.address.equals(address, ignoreCase = true) }
- ?.let {
+ val bondedDevice =
+ bluetoothRepository.state.value.bondedDevices.firstOrNull { it.address.equals(address, ignoreCase = true) }
+
+ if (bondedDevice != null) {
+ Logger.i { "[$address] Bonded device found; attempting short targeted scan for fresh advertisement" }
+ findTargetedDevice()?.let {
+ Logger.i { "[$address] Fresh advertisement found; using scanned device" }
return it
}
+ Logger.i { "[$address] Targeted scan timed out; falling back to bonded address" }
+ return bondedDevice
+ }
Logger.i { "[$address] Device not found in bonded list, scanning" }
@@ -210,6 +260,21 @@ class BleRadioTransport(
throw RadioNotConnectedException("Device not found at address $address")
}
+ private suspend fun findTargetedDevice(): BleDevice? = try {
+ withTimeoutOrNull(TARGETED_SCAN_TIMEOUT) {
+ // Pass both service UUID and address so the scanner can apply the most
+ // efficient platform filter while still keeping CoreBluetooth service-scoped.
+ scanner.scan(timeout = TARGETED_SCAN_TIMEOUT, serviceUuid = SERVICE_UUID, address = address).first {
+ it.address.equals(address, ignoreCase = true)
+ }
+ }
+ } catch (e: CancellationException) {
+ throw e
+ } catch (e: Exception) {
+ Logger.v(e) { "[$address] Targeted scan failed; falling back to bonded address" }
+ null
+ }
+
private fun connect() {
connectionJob =
connectionScope.launch {
@@ -226,12 +291,23 @@ class BleRadioTransport(
}
},
onTransientDisconnect = { error ->
- val msg = error?.toDisconnectReason()?.second ?: "Device unreachable"
- callback.onDisconnect(isPermanent = false, errorMessage = msg)
+ // Guard: if handleFailure already emitted the disconnect callback for this
+ // session (sessionFailed CAS won), don't emit a duplicate from the policy.
+ // Silent recovery: no errorMessage β the reconnect loop is still retrying, so
+ // a modal dialog would just confuse the user. The warning log is the
+ // observability surface for this transient event.
+ if (!sessionFailed.value) {
+ error?.let {
+ Logger.w(it) { "[$address] BLE reconnect attempt failed; continuing automatic retry" }
+ }
+ callback.onDisconnect(isPermanent = false)
+ }
},
onPermanentDisconnect = { error ->
- val msg = error?.toDisconnectReason()?.second ?: "Device unreachable"
- callback.onDisconnect(isPermanent = true, errorMessage = msg)
+ if (!sessionFailed.value) {
+ val msg = error?.toDisconnectReason()?.second ?: "Device unreachable"
+ callback.onDisconnect(isPermanent = true, errorMessage = msg)
+ }
},
)
}
@@ -243,9 +319,11 @@ class BleRadioTransport(
* Finds the device, bonds if needed, connects, discovers services, and waits for disconnect. Returns a
* [BleReconnectPolicy.Outcome] describing how the connection ended.
*/
- @Suppress("CyclomaticComplexMethod")
+ @Suppress("CyclomaticComplexMethod", "LongMethod", "ReturnCount")
private suspend fun attemptConnection(): BleReconnectPolicy.Outcome {
connectionStartTime = nowMillis
+ sessionFailed.value = false
+ sessionFailureCause = null
Logger.i { "[$address] BLE connection attempt started" }
val device = findDevice()
@@ -278,12 +356,49 @@ class BleRadioTransport(
discoverServicesAndSetupCharacteristics()
+ // If a fatal session failure (fromRadio/logRadio error) forced disconnect during setup,
+ // skip the Connected gate β return a retryable failure so BleReconnectPolicy handles it.
+ if (sessionFailureCause != null) {
+ Logger.w(sessionFailureCause) {
+ "[$address] Session failed during profile setup β returning failed outcome"
+ }
+ return BleReconnectPolicy.Outcome.Failed(sessionFailureCause ?: RuntimeException("Session setup failed"))
+ }
+
// Wait for the StateFlow to actually reflect Connected before watching for the next
// Disconnected. connectAndAwait returns synchronously based on the underlying Kable
// peripheral state, but our _connectionState observer runs on a separate coroutine and
// may lag. Without this gate the next .first { Disconnected } below could match the
// *previous* cycle's stale Disconnected value and fire immediately, breaking reconnect.
- bleConnection.connectionState.first { it is BleConnectionState.Connected }
+ //
+ // RACE GUARD: A fatal fromRadio/logRadio failure can land AFTER the sessionFailureCause
+ // check above but BEFORE connectionState reaches Connected. handleFailure() forces
+ // bleConnection.disconnect(), which should emit Disconnected β but if the StateFlow was
+ // already at a stale Disconnected value, it may NOT re-emit (StateFlow suppresses
+ // duplicate values). A bounded timeout ensures we cannot hang: if Connected doesn't
+ // arrive within CONNECTED_GATE_TIMEOUT, we check sessionFailureCause and return a
+ // retryable Failed outcome.
+ val connectedReached =
+ withTimeoutOrNull(CONNECTED_GATE_TIMEOUT) {
+ bleConnection.connectionState.first { it is BleConnectionState.Connected }
+ }
+ if (connectedReached == null) {
+ val failure = sessionFailureCause ?: RuntimeException("Timed out waiting for Connected state gate")
+ Logger.w(failure) { "[$address] Session failed before Connected gate β returning failed outcome" }
+ // CRITICAL: force GATT cleanup before returning Failed so we don't start a new
+ // attempt over an uncleared session. Without this, a timeout caused by flow lag or
+ // a stale-Disconnected state mismatch would leave a live/half-live GATT handle behind.
+ radioService = null
+ isFullyConnected = false
+ withContext(NonCancellable) {
+ try {
+ bleConnection.disconnect()
+ } catch (ignored: Exception) {
+ Logger.w(ignored) { "[$address] disconnect() failed during Connected-gate timeout cleanup" }
+ }
+ }
+ return BleReconnectPolicy.Outcome.Failed(failure)
+ }
// Suspend until the next Disconnected emission. We deliberately do NOT wrap this in a
// coroutineScope { launchIn(...); first(...) } pattern: launching a hot StateFlow
@@ -299,7 +414,19 @@ class BleRadioTransport(
Logger.i { "[$address] BLE connection dropped (reason: $disconnectReason), preparing to reconnect" }
- val wasIntentional = disconnectReason is DisconnectReason.LocalDisconnect
+ // Internal session failures (write/read exceptions that triggered handleFailure β
+ // disconnect) must NOT be treated as intentional/user disconnects β the reconnect policy
+ // needs to escalate backoff for these.
+ val internalFailure = sessionFailureCause
+ if (internalFailure != null) {
+ Logger.w(internalFailure) { "[$address] Session forced disconnect due to internal failure" }
+ }
+ val wasIntentional =
+ if (internalFailure != null) {
+ false
+ } else {
+ disconnectReason is DisconnectReason.LocalDisconnect
+ }
val connectionUptime = (nowMillis - gattConnectedAt).milliseconds
val wasStable = connectionUptime >= reconnectPolicy.minStableConnection
@@ -328,11 +455,21 @@ class BleRadioTransport(
private fun onDisconnected() {
radioService = null
+ // Atomic first-writer-wins: if handleFailure already claimed this session's failure
+ // callback (or another onDisconnected raced), CAS returns false and we skip the
+ // duplicate. The forced disconnect() from handleFailure causes Kable to emit
+ // Disconnected, which routes here β without this guard the UI would see two
+ // onDisconnect calls (one with the real error message from handleFailure, one
+ // generic from here).
+ val firstWriter = sessionFailed.compareAndSet(expect = false, update = true)
Logger.i { "[$address] BLE disconnected - ${formatSessionStats()}" }
- // Signal immediately so the UI reflects the disconnect while reconnect continues.
- callback.onDisconnect(isPermanent = false)
+ if (firstWriter) {
+ // Signal immediately so the UI reflects the disconnect while reconnect continues.
+ callback.onDisconnect(isPermanent = false)
+ }
}
+ @Suppress("LongMethod", "ThrowsCount")
private suspend fun discoverServicesAndSetupCharacteristics() {
try {
bleConnection.profile(serviceUuid = SERVICE_UUID) { service ->
@@ -365,7 +502,25 @@ class BleRadioTransport(
Logger.i { "[$address] Profile service active and characteristics subscribed" }
// Wait for FROMNUM CCCD write before triggering the Meshtastic handshake.
- radioService.awaitSubscriptionReady()
+ // Bounded: if fromRadio fails before subscriptionReady completes, handleFailure
+ // sets sessionFailureCause. The timeout also prevents a hang if FROMNUM observe
+ // never completes for reasons other than a fatal exception (e.g., firmware doesn't
+ // send CCCD confirmation). We MUST abort setup on timeout β proceeding without
+ // subscription readiness creates a half-initialized session.
+ val subscriptionReady =
+ withTimeoutOrNull(SUBSCRIPTION_READY_TIMEOUT) {
+ radioService.awaitSubscriptionReady()
+ true
+ } ?: false
+ if (!subscriptionReady || sessionFailed.value) {
+ val cause =
+ sessionFailureCause ?: RuntimeException("Timed out waiting for FROMNUM subscription readiness")
+ Logger.w(cause) {
+ val reason = if (!subscriptionReady) "timed out" else "failed"
+ "[$address] Subscription wait $reason β aborting setup"
+ }
+ throw cause
+ }
// Log negotiated MTU for diagnostics
val maxLen = bleConnection.maximumWriteValueLength(BleWriteType.WITHOUT_RESPONSE)
@@ -373,11 +528,38 @@ class BleRadioTransport(
requestHighPriorityAndScheduleDowngrade()
- this@BleRadioTransport.callback.onConnect()
+ // Guard: if handleFailure fired during setup (e.g., fromRadio/logRadio fatal
+ // exception after subscriptionReady completed but before this line), do NOT call
+ // onConnect β it would set Connected state on a dead session. handleFailure has
+ // already emitted the disconnect callback.
+ //
+ // ORDERING NOTE: callback.onConnect() is emitted here, BEFORE attemptConnection()
+ // re-confirms the link via the Connected gate (see CONNECTED_GATE_TIMEOUT). In the
+ // rare case the gate times out (connectionState observer-coroutine lag, or a stale
+ // Disconnected value the StateFlow does not re-emit), the UI has briefly seen
+ // Connected. This is deliberate and acceptable:
+ // - The gate-timeout path returns Outcome.Failed, which drives BleReconnectPolicy
+ // (Retry backoff immediately; onTransientDisconnect β DeviceSleep once
+ // consecutiveFailures reaches failureThreshold, default 3).
+ // - The timeout path nulls radioService and forces bleConnection.disconnect()
+ // under NonCancellable, so handleSendToRadio() fails fast against a null
+ // service and the next attempt starts over a clean GATT handle.
+ // The net worst case is a brief Connected indication while the transport cycles a
+ // sub-threshold retry β a cosmetic UX lag, not a correctness or data issue.
+ // Deferring onConnect until after the gate would require a structural refactor of
+ // the profile-setup callback and introduce its own races, so the current ordering
+ // is retained.
+ if (!sessionFailed.value) {
+ this@BleRadioTransport.callback.onConnect()
+ } else {
+ Logger.w { "[$address] Session failed during setup β skipping onConnect" }
+ }
}
} catch (e: CancellationException) {
// Scope was cancelled externally β still ensure GATT cleanup runs so we don't
// leak a BluetoothGatt handle and trigger GATT status 133 on the next attempt.
+ radioService = null
+ isFullyConnected = false
withContext(NonCancellable) {
try {
bleConnection.disconnect()
@@ -388,8 +570,10 @@ class BleRadioTransport(
throw e
} catch (e: Exception) {
Logger.w(e) { "[$address] Profile service discovery or operation failed" }
- // Disconnect to let the outer reconnect loop see a clean Disconnected state.
- // Do NOT call handleFailure here β the reconnect loop owns failure counting.
+ // Clear stale state so the next attempt starts clean β if failure happened after
+ // radioService assignment but before callback.onConnect(), stale state would survive.
+ radioService = null
+ isFullyConnected = false
withContext(NonCancellable) {
try {
bleConnection.disconnect()
@@ -397,6 +581,7 @@ class BleRadioTransport(
Logger.w(ignored) { "[$address] disconnect() failed after profile error" }
}
}
+ throw e // Re-throw so attemptConnection() returns Outcome.Failed(e) for policy backoff.
}
}
@@ -425,32 +610,56 @@ class BleRadioTransport(
/**
* Sends a packet to the radio with retry support.
*
+ * Write-failure policy: any non-cancellation exception from a failed write (after exhausting [retryBleOperation]
+ * retries) is treated as fatal to the current BLE session. Production logs show long-running sessions eventually
+ * failing writes with [NotConnectedException] after hundreds of successful writes β by that point the GATT handle
+ * is stale and retrying in-place cannot recover. Calling [handleFailure] forces a full GATT teardown + reconnect
+ * cycle, which is the only reliable recovery for a dead session. Transient single-write blips are absorbed by
+ * [retryBleOperation]'s 3-attempt retry before reaching this catch.
+ *
* @param p The packet to send.
*/
override fun handleSendToRadio(p: ByteArray) {
- val currentService = radioService
- if (currentService != null) {
- connectionScope.launch {
- writeMutex.withLock {
- try {
- retryBleOperation(tag = address) { currentService.sendToRadio(p) }
- packetsSent++
- bytesSent += p.size
- Logger.v {
- "[$address] Wrote packet #$packetsSent " +
- "to toRadio (${p.size} bytes, total TX: $bytesSent bytes)"
+ // Fast-path check: skip coroutine launch entirely if no transport is active.
+ if (radioService == null) {
+ Logger.w { "[$address] toRadio characteristic unavailable, can't send data" }
+ return
+ }
+ connectionScope.launch {
+ writeMutex.withLock {
+ // Re-read radioService UNDER the lock β handleFailure may have nulled it
+ // between the outer check and lock acquisition. Without this, a queued send
+ // can retry writes against a stale/dead profile.
+ val currentService =
+ radioService
+ ?: run {
+ Logger.w { "[$address] toRadio characteristic cleared during write queue" }
+ return@withLock
}
- } catch (e: Exception) {
+ try {
+ retryBleOperation(tag = address) { currentService.sendToRadio(p) }
+ val sent = packetsSent.incrementAndGet()
+ val txBytes = bytesSent.addAndGet(p.size.toLong())
+ Logger.v {
+ "[$address] Wrote packet #$sent " + "to toRadio (${p.size} bytes, total TX: $txBytes bytes)"
+ }
+ } catch (e: CancellationException) {
+ throw e
+ } catch (e: Exception) {
+ // Guard: only call handleFailure if this write was against the CURRENT session.
+ // If radioService was replaced (new reconnect cycle) or cleared (handleFailure
+ // already ran), this is a stale write from the old session β silently discard.
+ if (currentService === radioService) {
Logger.w(e) {
"[$address] Failed to write packet to toRadioCharacteristic after " +
- "$packetsSent successful writes"
+ "${packetsSent.value} successful writes"
}
handleFailure(e)
+ } else {
+ Logger.w(e) { "[$address] Stale write failure ignored (session was replaced)" }
}
}
}
- } else {
- Logger.w { "[$address] toRadio characteristic unavailable, can't send data" }
}
}
@@ -483,26 +692,68 @@ class BleRadioTransport(
}
private fun dispatchPacket(packet: ByteArray) {
- packetsReceived++
- bytesReceived += packet.size
- Logger.v {
- "[$address] Dispatching packet #$packetsReceived " +
- "(${packet.size} bytes, total RX: $bytesReceived bytes)"
- }
+ val received = packetsReceived.incrementAndGet()
+ val rxBytes = bytesReceived.addAndGet(packet.size.toLong())
+ Logger.v { "[$address] Dispatching packet #$received " + "(${packet.size} bytes, total RX: $rxBytes bytes)" }
callback.handleFromRadio(packet)
}
+ /**
+ * Preserves the first session-failure cause across concurrent failures. Called before the [sessionFailed] CAS in
+ * [handleFailure] and [exceptionHandler] to eliminate the window where [sessionFailed] is true but
+ * [sessionFailureCause] is still null.
+ */
+ private fun recordSessionFailureCause(throwable: Throwable) {
+ if (sessionFailureCause == null) sessionFailureCause = throwable
+ }
+
private fun handleFailure(throwable: Throwable) {
+ // CancellationException signals intentional scope cancellation (close() called).
+ // Never surface it as a user-facing disconnect error.
+ if (throwable is CancellationException) return
+
+ // Record the cause BEFORE the CAS so there is no window where another coroutine
+ // sees sessionFailed == true but sessionFailureCause == null. first-cause is
+ // preserved: a concurrent loser only overwrites if still null.
+ recordSessionFailureCause(throwable)
+
+ // Deduplicate via atomic CAS: only the first failure per connection attempt triggers
+ // session teardown. Heartbeat writes that arrive after the first failure must not spam
+ // callbacks. compareAndSet(false, true) returns true iff THIS caller is the first.
+ if (!sessionFailed.compareAndSet(expect = false, update = true)) return
+
+ // Tear down stale session state immediately so future writes fail-fast without retrying
+ // against a dead GATT handle.
+ radioService = null
+ isFullyConnected = false
+
val (isPermanent, msg) = throwable.toDisconnectReason()
- callback.onDisconnect(isPermanent, errorMessage = msg)
+ // Silent recovery for non-permanent failures: the transport tears down stale GATT state
+ // and reconnects automatically, so surfacing a modal for a transient session failure is
+ // confusing UX. Permanent failures (pairing, missing characteristic, etc.) remain
+ // user-facing.
+ callback.onDisconnect(isPermanent, errorMessage = if (isPermanent) msg else null)
+
+ Logger.w(throwable) { "[$address] Session failure β forcing cleanup for reconnect" }
+
+ // Force GATT disconnect on the detached cleanupScope (matching the pattern used by
+ // the exceptionHandler defined above). This causes Kable's connectionState
+ // to emit Disconnected, unblocking attemptConnection so BleReconnectPolicy iterates.
+ cleanupScope.launch {
+ try {
+ bleConnection.disconnect()
+ } catch (e: Exception) {
+ Logger.w(e) { "[$address] Failed to disconnect after session failure" }
+ }
+ }
}
/** Formats a one-line session statistics summary for logging. */
private fun formatSessionStats(): String {
val uptime = if (connectionStartTime > 0) nowMillis - connectionStartTime else 0
return "Uptime: ${uptime}ms, " +
- "Packets RX: $packetsReceived ($bytesReceived bytes), " +
- "Packets TX: $packetsSent ($bytesSent bytes)"
+ "Packets RX: ${packetsReceived.value} (${bytesReceived.value} bytes), " +
+ "Packets TX: ${packetsSent.value} (${bytesSent.value} bytes)"
}
private fun Throwable.toDisconnectReason(): Pair<Boolean, String> {
diff --git a/core/network/src/commonTest/kotlin/org/meshtastic/core/network/radio/BleRadioTransportReconnectCrashTest.kt b/core/network/src/commonTest/kotlin/org/meshtastic/core/network/radio/BleRadioTransportReconnectCrashTest.kt
index c1835e7881..6bd4bb486a 100644
--- a/core/network/src/commonTest/kotlin/org/meshtastic/core/network/radio/BleRadioTransportReconnectCrashTest.kt
+++ b/core/network/src/commonTest/kotlin/org/meshtastic/core/network/radio/BleRadioTransportReconnectCrashTest.kt
@@ -16,7 +16,10 @@
*/
package org.meshtastic.core.network.radio
+import com.juul.kable.GattStatusException
+import com.juul.kable.NotConnectedException
import dev.mokkery.MockMode
+import dev.mokkery.answering.calls
import dev.mokkery.answering.returns
import dev.mokkery.every
import dev.mokkery.matcher.any
@@ -24,6 +27,7 @@ import dev.mokkery.mock
import kotlinx.coroutines.CancellationException
import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.ExperimentalCoroutinesApi
+import kotlinx.coroutines.currentCoroutineContext
import kotlinx.coroutines.flow.MutableStateFlow
import kotlinx.coroutines.flow.StateFlow
import kotlinx.coroutines.flow.asStateFlow
@@ -36,13 +40,18 @@ import org.meshtastic.core.ble.BleDevice
import org.meshtastic.core.ble.BleService
import org.meshtastic.core.ble.BleWriteType
import org.meshtastic.core.ble.DisconnectReason
+import org.meshtastic.core.ble.MeshtasticBleConstants.FROMNUM_CHARACTERISTIC
+import org.meshtastic.core.ble.MeshtasticBleConstants.FROMRADIO_CHARACTERISTIC
+import org.meshtastic.core.ble.MeshtasticBleConstants.SERVICE_UUID
import org.meshtastic.core.testing.FakeBleConnection
import org.meshtastic.core.testing.FakeBleConnectionFactory
import org.meshtastic.core.testing.FakeBleDevice
import org.meshtastic.core.testing.FakeBleScanner
+import org.meshtastic.core.testing.FakeBleService
import org.meshtastic.core.testing.FakeBluetoothRepository
import kotlin.test.BeforeTest
import kotlin.test.Test
+import kotlin.test.assertEquals
import kotlin.test.assertTrue
import kotlin.time.Duration
@@ -86,6 +95,7 @@ class BleRadioTransportReconnectCrashTest {
fun `close calls disconnect to clean up GATT handle`() = runTest {
val device = FakeBleDevice(address = address, name = "Test Radio")
bluetoothRepository.bond(device)
+ scanner.emitDevice(device)
val bleTransport =
BleRadioTransport(
@@ -96,16 +106,20 @@ class BleRadioTransportReconnectCrashTest {
callback = service,
address = address,
)
- bleTransport.start()
+ try {
+ bleTransport.start()
- // Allow the connection loop to reach the connected state.
- advanceTimeBy(4_000L)
+ // Allow the connection loop to reach the connected state.
+ advanceTimeBy(4_000L)
- bleTransport.close()
+ bleTransport.close()
- // disconnect() must be called: once by the connection loop teardown + once by close() itself.
- // We only assert it was called at least once β the exact count depends on timing.
- assertTrue(connection.disconnectCalls >= 1, "Expected disconnect() to be called at least once")
+ // disconnect() must be called: once by the connection loop teardown + once by close() itself.
+ // We only assert it was called at least once β the exact count depends on timing.
+ assertTrue(connection.disconnectCalls >= 1, "Expected disconnect() to be called at least once")
+ } finally {
+ bleTransport.close()
+ }
}
// βββ disconnect called on connection failure ββββββββββββββββββββββββββββββββββββββββββββββββββ
@@ -121,6 +135,7 @@ class BleRadioTransportReconnectCrashTest {
fun `disconnect is called on connection failure`() = runTest {
val device = FakeBleDevice(address = address, name = "Test Radio")
bluetoothRepository.bond(device)
+ scanner.emitDevice(device)
// Make every connection attempt fail.
connection.failNextN = Int.MAX_VALUE
@@ -134,14 +149,18 @@ class BleRadioTransportReconnectCrashTest {
callback = service,
address = address,
)
- bleTransport.start()
+ try {
+ bleTransport.start()
- advanceTimeBy(30_000L)
+ advanceTimeBy(30_000L)
- bleTransport.close()
+ bleTransport.close()
- // Each failed connectAndAwait round-trips through the reconnect loop; close() always disconnects.
- assertTrue(connection.disconnectCalls >= 1, "disconnect() not called after connection failure")
+ // Each failed connectAndAwait round-trips through the reconnect loop; close() always disconnects.
+ assertTrue(connection.disconnectCalls >= 1, "disconnect() not called after connection failure")
+ } finally {
+ bleTransport.close()
+ }
}
// βββ transient onDisconnect after failure threshold ββββββββββββββββββββββββββββββββββββββββββ
@@ -157,6 +176,7 @@ class BleRadioTransportReconnectCrashTest {
fun `transient onDisconnect is signalled after failure threshold without giving up`() = runTest {
val device = FakeBleDevice(address = address, name = "Test Radio")
bluetoothRepository.bond(device)
+ scanner.emitDevice(device)
connection.connectException = org.meshtastic.core.model.RadioNotConnectedException("simulated GATT failure")
@@ -171,18 +191,23 @@ class BleRadioTransportReconnectCrashTest {
callback = service,
address = address,
)
- bleTransport.start()
+ try {
+ bleTransport.start()
- advanceTimeBy(24_001L)
+ advanceTimeBy(24_001L)
- // Transient disconnect must have been signalled.
- dev.mokkery.verify { service.onDisconnect(isPermanent = false, errorMessage = any()) }
- // Permanent disconnect must NEVER be called by the transport on its own.
- dev.mokkery.verify(mode = dev.mokkery.verify.VerifyMode.not) {
- service.onDisconnect(isPermanent = true, errorMessage = any())
- }
+ // Transient disconnect must be signalled with NO user-facing error message β the
+ // reconnect loop is still retrying, so a modal dialog would be confusing UX.
+ dev.mokkery.verify { service.onDisconnect(isPermanent = false, errorMessage = null) }
+ // Permanent disconnect must NEVER be called by the transport on its own.
+ dev.mokkery.verify(mode = dev.mokkery.verify.VerifyMode.not) {
+ service.onDisconnect(isPermanent = true, errorMessage = any())
+ }
- bleTransport.close()
+ bleTransport.close()
+ } finally {
+ bleTransport.close()
+ }
}
// βββ CancellationException is not silently swallowed βββββββββββββββββββββββββββββββββββββββββ
@@ -206,6 +231,7 @@ class BleRadioTransportReconnectCrashTest {
}
val device = FakeBleDevice(address = address, name = "Test Radio")
bluetoothRepository.bond(device)
+ scanner.emitDevice(device)
val bleTransport =
BleRadioTransport(
@@ -216,17 +242,21 @@ class BleRadioTransportReconnectCrashTest {
callback = service,
address = address,
)
- bleTransport.start()
+ try {
+ bleTransport.start()
- // Allow one connection attempt to reach profile() and be cancelled.
- advanceTimeBy(4_000L)
+ // Allow one connection attempt to reach profile() and be cancelled.
+ advanceTimeBy(4_000L)
- bleTransport.close()
+ bleTransport.close()
- assertTrue(
- throwingConnection.disconnectCalls >= 1,
- "disconnect() must be called after CancellationException in profile() β GATT leak fix",
- )
+ assertTrue(
+ throwingConnection.disconnectCalls >= 1,
+ "disconnect() must be called after CancellationException in profile() β GATT leak fix",
+ )
+ } finally {
+ bleTransport.close()
+ }
}
// βββ Reconnect after a stable connection drops βββββββββββββββββββββββββββββββββββββββββββββββ
@@ -251,6 +281,509 @@ class BleRadioTransportReconnectCrashTest {
fun `transport reconnects after a stable connection is dropped remotely`() = runTest {
val device = FakeBleDevice(address = address, name = "Test Radio")
bluetoothRepository.bond(device)
+ scanner.emitDevice(device)
+
+ val bleTransport =
+ BleRadioTransport(
+ scope = this,
+ scanner = scanner,
+ bluetoothRepository = bluetoothRepository,
+ connectionFactory = connectionFactory,
+ callback = service,
+ address = address,
+ )
+ try {
+ bleTransport.start()
+
+ // Settle delay (3 s) + connect + handshake.
+ advanceTimeBy(4_000L)
+ assertTrue(connection.connectAndAwaitCalls == 1, "First connect must happen during initial start window")
+
+ // Stay connected long enough to be considered stable (> minStableConnection = 5 s).
+ advanceTimeBy(10_000L)
+
+ // Simulate the firmware dying mid-session β the same path a node power-cycle takes.
+ connection.simulateRemoteDisconnect(reason = DisconnectReason.Timeout)
+
+ // Settle delay (3 s) before the next attempt + re-connect window. Generous to absorb
+ // the policy retry backoff (5 s on first failure) plus another 3 s settle delay.
+ advanceTimeBy(30_000L)
+
+ assertTrue(
+ connection.connectAndAwaitCalls >= 2,
+ "Reconnect loop must call connectAndAwait again after a remote disconnect " +
+ "(actual calls: ${connection.connectAndAwaitCalls})",
+ )
+
+ bleTransport.close()
+ } finally {
+ bleTransport.close()
+ }
+ }
+
+ // βββ Session-failure recovery ββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ
+
+ @Test
+ fun `write failure while connected clears state and triggers reconnect`() = runTest {
+ val device = FakeBleDevice(address = address, name = "Test Radio")
+ bluetoothRepository.bond(device)
+ scanner.emitDevice(device)
+
+ val bleTransport =
+ BleRadioTransport(
+ scope = this,
+ scanner = scanner,
+ bluetoothRepository = bluetoothRepository,
+ connectionFactory = connectionFactory,
+ callback = service,
+ address = address,
+ )
+ try {
+ bleTransport.start()
+ advanceTimeBy(4_000L) // connect + handshake
+ assertTrue(connection.connectAndAwaitCalls == 1, "First connect must happen")
+
+ // Inject write failure while BLE state remains Connected
+ connection.service.writeException = NotConnectedException("session closed")
+
+ // Trigger a write (simulating a heartbeat or user packet)
+ bleTransport.handleSendToRadio(byteArrayOf(1, 2, 3))
+ // Advance through retryBleOperation's 3 retries (~750ms backoff) so the write failure
+ // reaches handleFailure and forces disconnect. Stay under the 3s reconnect settle delay.
+ advanceTimeBy(1_000L)
+
+ // After failure: disconnect must be called (GATT cleanup)
+ assertTrue(connection.disconnectCalls >= 1, "disconnect() must be called after write failure")
+
+ // Reconnect policy should iterate β wait for settle (3s) + connect
+ advanceTimeBy(10_000L)
+ assertTrue(
+ connection.connectAndAwaitCalls >= 2,
+ "Reconnect loop must call connectAndAwait again after session failure " +
+ "(actual calls: ${connection.connectAndAwaitCalls})",
+ )
+
+ bleTransport.close()
+ } finally {
+ bleTransport.close()
+ }
+ }
+
+ @Test
+ fun `CancellationException from write does not trigger callback`() = runTest {
+ val device = FakeBleDevice(address = address, name = "Test Radio")
+ bluetoothRepository.bond(device)
+ scanner.emitDevice(device)
+
+ every { service.onDisconnect(any(), any()) } returns Unit
+
+ val bleTransport =
+ BleRadioTransport(
+ scope = this,
+ scanner = scanner,
+ bluetoothRepository = bluetoothRepository,
+ connectionFactory = connectionFactory,
+ callback = service,
+ address = address,
+ )
+ try {
+ bleTransport.start()
+ advanceTimeBy(4_000L)
+
+ connection.service.writeException = CancellationException("cancelled")
+
+ bleTransport.handleSendToRadio(byteArrayOf(1))
+ // Drain the cancelled write coroutine; no retry is triggered for CancellationException.
+ testScheduler.runCurrent()
+
+ // CancellationException must NOT result in a user-facing disconnect callback
+ dev.mokkery.verify(mode = dev.mokkery.verify.VerifyMode.not) { service.onDisconnect(any(), any()) }
+
+ bleTransport.close()
+ } finally {
+ bleTransport.close()
+ }
+ }
+
+ @Test
+ fun `repeated writes after session failure do not spam onDisconnect`() = runTest {
+ val device = FakeBleDevice(address = address, name = "Test Radio")
+ bluetoothRepository.bond(device)
+ scanner.emitDevice(device)
+
+ var onDisconnectCalls = 0
+ every { service.onDisconnect(any(), any()) } calls { onDisconnectCalls++ }
+
+ val bleTransport =
+ BleRadioTransport(
+ scope = this,
+ scanner = scanner,
+ bluetoothRepository = bluetoothRepository,
+ connectionFactory = connectionFactory,
+ callback = service,
+ address = address,
+ )
+ try {
+ bleTransport.start()
+ advanceTimeBy(4_000L)
+
+ // First write failure β should trigger onDisconnect once
+ connection.service.writeException = NotConnectedException("session closed")
+ bleTransport.handleSendToRadio(byteArrayOf(1))
+ // Advance enough for retryBleOperation to exhaust 3 retries (~750ms max) + handleFailure +
+ // disconnect, but NOT enough to reach the 3 s reconnect settle delay.
+ advanceTimeBy(2_000L)
+
+ // Second write failure against same session β must NOT trigger another callback
+ bleTransport.handleSendToRadio(byteArrayOf(2))
+ advanceTimeBy(1_000L)
+
+ assertEquals(1, onDisconnectCalls, "onDisconnect must be called exactly once for the session failure")
+
+ bleTransport.close()
+ } finally {
+ bleTransport.close()
+ }
+ }
+
+ @Test
+ fun `stale radioService is cleared after session failure`() = runTest {
+ val device = FakeBleDevice(address = address, name = "Test Radio")
+ bluetoothRepository.bond(device)
+ scanner.emitDevice(device)
+
+ val bleTransport =
+ BleRadioTransport(
+ scope = this,
+ scanner = scanner,
+ bluetoothRepository = bluetoothRepository,
+ connectionFactory = connectionFactory,
+ callback = service,
+ address = address,
+ )
+ try {
+ bleTransport.start()
+ advanceTimeBy(4_000L)
+
+ val writesBefore = connection.service.writes.size
+
+ // First write failure clears radioService
+ connection.service.writeException = NotConnectedException("session closed")
+ bleTransport.handleSendToRadio(byteArrayOf(1, 2, 3))
+ // Advance enough for retryBleOperation to exhaust 3 retries (~750ms max) + handleFailure +
+ // disconnect, but NOT enough to reach the 3 s reconnect settle delay.
+ advanceTimeBy(2_000L)
+
+ // Second write β radioService should be null, so no write is attempted
+ connection.service.writeException = null // clear the exception hook
+ bleTransport.handleSendToRadio(byteArrayOf(4, 5, 6))
+ advanceTimeBy(1_000L)
+
+ // No new writes should have been recorded (radioService was null β write skipped)
+ assertEquals(
+ writesBefore,
+ connection.service.writes.size,
+ "No new write should be recorded after radioService was cleared",
+ )
+
+ bleTransport.close()
+ } finally {
+ bleTransport.close()
+ }
+ }
+
+ @Test
+ fun `internal session failure is not treated as intentional disconnect`() = runTest {
+ val device = FakeBleDevice(address = address, name = "Test Radio")
+ bluetoothRepository.bond(device)
+ scanner.emitDevice(device)
+
+ every { service.onDisconnect(any(), any()) } returns Unit
+
+ val bleTransport =
+ BleRadioTransport(
+ scope = this,
+ scanner = scanner,
+ bluetoothRepository = bluetoothRepository,
+ connectionFactory = connectionFactory,
+ callback = service,
+ address = address,
+ )
+ try {
+ bleTransport.start()
+ advanceTimeBy(4_000L) // connect + handshake
+
+ // Inject a write failure β this should NOT be treated as intentional
+ connection.service.writeException = NotConnectedException("session closed")
+ bleTransport.handleSendToRadio(byteArrayOf(1, 2, 3))
+ // Advance through retryBleOperation's 3 retries (~750ms backoff) so the write failure
+ // reaches handleFailure and forces disconnect.
+ advanceTimeBy(1_000L)
+
+ // disconnect must have been called (forced cleanup)
+ assertTrue(connection.disconnectCalls >= 1, "disconnect() must be called after write failure")
+
+ // Verify onDisconnect was called with isPermanent = false and NO user-facing error β
+ // non-permanent session failures auto-recover, so no modal dialog should be shown.
+ dev.mokkery.verify { service.onDisconnect(isPermanent = false, errorMessage = null) }
+ dev.mokkery.verify(mode = dev.mokkery.verify.VerifyMode.not) {
+ service.onDisconnect(isPermanent = true, errorMessage = any())
+ }
+
+ // Reconnect should happen (policy should iterate)
+ advanceTimeBy(15_000L)
+ assertTrue(
+ connection.connectAndAwaitCalls >= 2,
+ "Reconnect loop must iterate after internal session failure",
+ )
+
+ bleTransport.close()
+ } finally {
+ bleTransport.close()
+ }
+ }
+
+ // βββ Liveness restart semantics ββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ
+
+ /**
+ * Validates the transport-level behavior that liveness restart depends on: after stop (close) + start, a new
+ * connection attempt must occur.
+ *
+ * The [SharedRadioInterfaceService.checkLiveness] path calls stopTransportLocked then startTransportLocked, which
+ * destroys and recreates the transport. This test verifies that BleRadioTransport correctly handles this stop/start
+ * cycle.
+ */
+ @Test
+ fun `stop and restart creates a new connection attempt`() = runTest {
+ val device = FakeBleDevice(address = address, name = "Test Radio")
+ bluetoothRepository.bond(device)
+ scanner.emitDevice(device)
+
+ val bleTransport =
+ BleRadioTransport(
+ scope = this,
+ scanner = scanner,
+ bluetoothRepository = bluetoothRepository,
+ connectionFactory = connectionFactory,
+ callback = service,
+ address = address,
+ )
+ try {
+ bleTransport.start()
+ advanceTimeBy(4_000L)
+ assertTrue(connection.connectAndAwaitCalls == 1, "First connect must happen")
+
+ // Simulate liveness restart: close + create fresh transport
+ bleTransport.close()
+ advanceTimeBy(2_000L)
+
+ val freshConnection = FakeBleConnection()
+ val freshFactory = FakeBleConnectionFactory(freshConnection)
+
+ val restartedTransport =
+ BleRadioTransport(
+ scope = this,
+ scanner = scanner,
+ bluetoothRepository = bluetoothRepository,
+ connectionFactory = freshFactory,
+ callback = service,
+ address = address,
+ )
+ try {
+ restartedTransport.start()
+ advanceTimeBy(4_000L)
+
+ assertTrue(
+ freshConnection.connectAndAwaitCalls >= 1,
+ "Fresh transport must attempt connection after restart",
+ )
+
+ restartedTransport.close()
+ } finally {
+ restartedTransport.close()
+ }
+ } finally {
+ bleTransport.close()
+ }
+ }
+
+ // βββ Profile setup failure returns Failed outcome and retries ββββββββββββββββββββββββββββββββββ
+
+ /**
+ * When profile setup fails (e.g. missing service UUID), the transport should return a
+ * [BleReconnectPolicy.Outcome.Failed] and the reconnect loop should iterate. Uses
+ * [FakeBleConnection.missingServices] to cause `profile()` to throw.
+ */
+ @Test
+ fun `profile setup failure returns Failed outcome and retries`() = runTest {
+ val device = FakeBleDevice(address = address, name = "Test Radio")
+ bluetoothRepository.bond(device)
+ scanner.emitDevice(device)
+
+ // Make SERVICE_UUID missing β profile() throws NoSuchElementException
+ connection.missingServices.add(SERVICE_UUID)
+
+ val bleTransport =
+ BleRadioTransport(
+ scope = this,
+ scanner = scanner,
+ bluetoothRepository = bluetoothRepository,
+ connectionFactory = connectionFactory,
+ callback = service,
+ address = address,
+ )
+ try {
+ bleTransport.start()
+
+ // First attempt fails at profile setup, then settle delay + reconnect.
+ advanceTimeBy(15_000L)
+
+ // The reconnect loop should have called connectAndAwait at least twice
+ // (initial failure + retry)
+ assertTrue(
+ connection.connectAndAwaitCalls >= 2,
+ "Reconnect loop must iterate after profile setup failure " +
+ "(actual calls: ${connection.connectAndAwaitCalls})",
+ )
+
+ bleTransport.close()
+ } finally {
+ bleTransport.close()
+ }
+ }
+
+ /**
+ * Regression: CancellationException from a write should not trigger onDisconnect even after the transport has
+ * reconnected and is operating normally.
+ */
+ @Test
+ fun `CancellationException from write does not trigger onDisconnect after reconnection`() = runTest {
+ val device = FakeBleDevice(address = address, name = "Test Radio")
+ bluetoothRepository.bond(device)
+ scanner.emitDevice(device)
+
+ every { service.onDisconnect(any(), any()) } returns Unit
+
+ val bleTransport =
+ BleRadioTransport(
+ scope = this,
+ scanner = scanner,
+ bluetoothRepository = bluetoothRepository,
+ connectionFactory = connectionFactory,
+ callback = service,
+ address = address,
+ )
+ try {
+ bleTransport.start()
+ advanceTimeBy(4_000L)
+
+ // Inject CancellationException into a write
+ connection.service.writeException = CancellationException("cancelled")
+ bleTransport.handleSendToRadio(byteArrayOf(1))
+ // Drain the cancelled write coroutine; no retry is triggered for CancellationException.
+ testScheduler.runCurrent()
+
+ // CancellationException must NOT result in a user-facing disconnect callback
+ dev.mokkery.verify(mode = dev.mokkery.verify.VerifyMode.not) { service.onDisconnect(any(), any()) }
+
+ // Clear the exception and verify the transport is still usable
+ connection.service.writeException = null
+ bleTransport.handleSendToRadio(byteArrayOf(2))
+ advanceTimeBy(1_000L)
+
+ // Still no disconnect should have been called
+ dev.mokkery.verify(mode = dev.mokkery.verify.VerifyMode.not) { service.onDisconnect(any(), any()) }
+
+ bleTransport.close()
+ } finally {
+ bleTransport.close()
+ }
+ }
+
+ // βββ Read-path session-failure recovery (fromRadio fatal) ββββββββββββββββββββββββββββββββββββββ
+
+ /**
+ * Validates that a session-fatal exception in the fromRadio READ path (not just the write path) triggers
+ * handleFailure β forced disconnect β reconnect.
+ *
+ * Existing tests only exercise the WRITE path. The read-path recovery is the primary mechanism for detecting zombie
+ * sessions where fromRadio throws GATT 133/19/8/129 during a poll.
+ */
+ @Test
+ fun `fromRadio read failure triggers handleFailure and reconnect`() = runTest {
+ val device = FakeBleDevice(address = address, name = "Test Radio")
+ bluetoothRepository.bond(device)
+ scanner.emitDevice(device)
+
+ // Register FROMNUM + FROMRADIO before start() so the profile's observe collector and read
+ // loop are set up during discoverServicesAndSetupCharacteristics. Without FROMNUM, no
+ // notification collector is established (emitNotification fires into a void). Without
+ // FROMRADIO, hasCharacteristic(fromRadioChar) returns false and the read loop skips before
+ // readException can fire.
+ connection.service.addCharacteristic(FROMNUM_CHARACTERISTIC)
+ connection.service.addCharacteristic(FROMRADIO_CHARACTERISTIC)
+
+ val bleTransport =
+ BleRadioTransport(
+ scope = this,
+ scanner = scanner,
+ bluetoothRepository = bluetoothRepository,
+ connectionFactory = connectionFactory,
+ callback = service,
+ address = address,
+ )
+ try {
+ bleTransport.start()
+ advanceTimeBy(4_000L) // connect + handshake
+ assertTrue(connection.connectAndAwaitCalls == 1, "First connect must happen")
+
+ // Inject a read failure that will fire on the next drain cycle
+ connection.service.readException = NotConnectedException("read session closed")
+
+ // Trigger a drain by emitting a FROMNUM notification β this causes the profile to poll
+ // fromRadioChar, which throws NotConnectedException (a session-fatal BLE exception).
+ connection.service.emitNotification(FROMNUM_CHARACTERISTIC, byteArrayOf(1))
+ // Drain the immediate fromRadio failure and forced disconnect; the explicit advanceTimeBy
+ // below covers the reconnect loop's delayed retry.
+ testScheduler.runCurrent()
+
+ // handleFailure must have forced a GATT disconnect
+ assertTrue(connection.disconnectCalls >= 1, "disconnect() must be called after fromRadio read failure")
+
+ // Reconnect policy should iterate
+ advanceTimeBy(10_000L)
+ assertTrue(
+ connection.connectAndAwaitCalls >= 2,
+ "Reconnect loop must iterate after fromRadio read failure " +
+ "(actual calls: ${connection.connectAndAwaitCalls})",
+ )
+
+ bleTransport.close()
+ } finally {
+ bleTransport.close()
+ }
+ }
+
+ /**
+ * Validates the sessionFailed dedup guard when read-path and write-path failures race.
+ *
+ * The guard at the top of handleFailure exists precisely for this scenario: fromRadio's .catch handler and a
+ * concurrent write failure both call handleFailure against the same dead session. Only the first caller should fire
+ * onDisconnect.
+ */
+ @Test
+ fun `concurrent fromRadio and write failure fires onDisconnect exactly once`() = runTest {
+ val device = FakeBleDevice(address = address, name = "Test Radio")
+ bluetoothRepository.bond(device)
+ scanner.emitDevice(device)
+
+ // Register FROMNUM + FROMRADIO before start() so both read and write failure paths can
+ // fire. See the read-failure test above for details.
+ connection.service.addCharacteristic(FROMNUM_CHARACTERISTIC)
+ connection.service.addCharacteristic(FROMRADIO_CHARACTERISTIC)
+
+ var onDisconnectCalls = 0
+ every { service.onDisconnect(any(), any()) } calls { onDisconnectCalls++ }
val bleTransport =
BleRadioTransport(
@@ -261,29 +794,236 @@ class BleRadioTransportReconnectCrashTest {
callback = service,
address = address,
)
- bleTransport.start()
+ try {
+ bleTransport.start()
+ advanceTimeBy(4_000L)
- // Settle delay (3 s) + connect + handshake.
- advanceTimeBy(4_000L)
- assertTrue(connection.connectAndAwaitCalls == 1, "First connect must happen during initial start window")
+ // Inject BOTH read and write failures against the same session
+ connection.service.readException = NotConnectedException("read failure")
+ connection.service.writeException = NotConnectedException("write failure")
- // Stay connected long enough to be considered stable (> minStableConnection = 5 s).
- advanceTimeBy(10_000L)
+ // Trigger both paths nearly simultaneously
+ connection.service.emitNotification(FROMNUM_CHARACTERISTIC, byteArrayOf(1))
+ bleTransport.handleSendToRadio(byteArrayOf(42))
+ // Advance through the write-path retryBleOperation (~750ms) so BOTH the read and write
+ // failure paths reach handleFailure, truly exercising the CAS dedup guard.
+ advanceTimeBy(1_000L)
- // Simulate the firmware dying mid-session β the same path a node power-cycle takes.
- connection.simulateRemoteDisconnect(reason = DisconnectReason.Timeout)
+ assertEquals(
+ 1,
+ onDisconnectCalls,
+ "onDisconnect must fire exactly once despite concurrent read+write failures " +
+ "(actual: $onDisconnectCalls)",
+ )
+
+ bleTransport.close()
+ } finally {
+ bleTransport.close()
+ }
+ }
- // Settle delay (3 s) before the next attempt + re-connect window. Generous to absorb
- // the policy retry backoff (5 s on first failure) plus another 3 s settle delay.
- advanceTimeBy(30_000L)
+ // βββ FROMNUM subscription readiness setup failures βββββββββββββββββββββββββββββββββββββββββββ
- assertTrue(
- connection.connectAndAwaitCalls >= 2,
- "Reconnect loop must call connectAndAwait again after a remote disconnect " +
- "(actual calls: ${connection.connectAndAwaitCalls})",
+ @Test
+ fun `FROMNUM NotConnected before readiness aborts setup and retries`() =
+ fromNumPreReadinessFailureAbortsSetupAndRetries(
+ failure = NotConnectedException("FROMNUM observe failed before CCCD"),
)
- bleTransport.close()
+ @Test
+ fun `FROMNUM GATT 133 before readiness aborts setup and retries`() =
+ fromNumPreReadinessFailureAbortsSetupAndRetries(
+ failure = GattStatusException(status = 133, message = "GATT error"),
+ )
+
+ private fun fromNumPreReadinessFailureAbortsSetupAndRetries(failure: Exception) = runTest {
+ val device = FakeBleDevice(address = address, name = "Test Radio")
+ bluetoothRepository.bond(device)
+ scanner.emitDevice(device)
+
+ connection.service.addCharacteristic(FROMNUM_CHARACTERISTIC)
+ connection.service.observeBeforeSubscriptionExceptionByCharacteristic[FROMNUM_CHARACTERISTIC] = failure
+
+ var onConnectCalls = 0
+ every { service.onConnect() } calls { onConnectCalls++ }
+ every { service.onDisconnect(any(), any()) } returns Unit
+
+ val bleTransport =
+ BleRadioTransport(
+ scope = this,
+ scanner = scanner,
+ bluetoothRepository = bluetoothRepository,
+ connectionFactory = connectionFactory,
+ callback = service,
+ address = address,
+ )
+ try {
+ bleTransport.start()
+
+ // Initial settle (3s) + setup failure. Assert before retry can succeed, proving the failed
+ // pre-readiness attempt did not falsely mark subscription readiness or call onConnect().
+ advanceTimeBy(4_000L)
+
+ assertTrue(connection.disconnectCalls >= 1, "disconnect() must be called after FROMNUM observe failure")
+ assertEquals(0, onConnectCalls, "Failed pre-readiness setup must not call onConnect")
+
+ advanceTimeBy(10_000L)
+ assertTrue(
+ connection.connectAndAwaitCalls >= 2,
+ "Reconnect loop must retry after FROMNUM pre-readiness failure " +
+ "(actual calls: ${connection.connectAndAwaitCalls})",
+ )
+
+ bleTransport.close()
+ } finally {
+ bleTransport.close()
+ }
+ }
+
+ @Test
+ fun `FROMNUM subscription readiness timeout aborts setup clears stale service and retries`() = runTest {
+ val device = FakeBleDevice(address = address, name = "Test Radio")
+ bluetoothRepository.bond(device)
+ scanner.emitDevice(device)
+
+ connection.service.addCharacteristic(FROMNUM_CHARACTERISTIC)
+ connection.service.observeNeverSubscribeCharacteristics += FROMNUM_CHARACTERISTIC
+
+ var onConnectCalls = 0
+ every { service.onConnect() } calls { onConnectCalls++ }
+ every { service.onDisconnect(any(), any()) } returns Unit
+
+ val bleTransport =
+ BleRadioTransport(
+ scope = this,
+ scanner = scanner,
+ bluetoothRepository = bluetoothRepository,
+ connectionFactory = connectionFactory,
+ callback = service,
+ address = address,
+ )
+ try {
+ bleTransport.start()
+
+ // Settle (3s) + SUBSCRIPTION_READY_TIMEOUT (5s) + margin. FROMNUM observe never invokes
+ // onSubscription, so setup must abort instead of continuing with a half-initialized service.
+ advanceTimeBy(9_000L)
+
+ assertTrue(connection.disconnectCalls >= 1, "disconnect() must be called after subscription timeout")
+ assertEquals(0, onConnectCalls, "Subscription timeout must not call onConnect")
+
+ val writesBefore = connection.service.writes.size
+ bleTransport.handleSendToRadio(byteArrayOf(7, 8, 9))
+ assertEquals(
+ writesBefore,
+ connection.service.writes.size,
+ "Timed-out setup must clear radioService so writes do not use a half-initialized service",
+ )
+
+ advanceTimeBy(10_000L)
+ assertTrue(
+ connection.connectAndAwaitCalls >= 2,
+ "Reconnect loop must retry after subscription readiness timeout " +
+ "(actual calls: ${connection.connectAndAwaitCalls})",
+ )
+
+ bleTransport.close()
+ } finally {
+ bleTransport.close()
+ }
+ }
+
+ // βββ Connected-gate timeout cleanup ββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ
+
+ @Test
+ fun `Connected-gate timeout cleans up stale service and retries without disconnect spam`() = runTest {
+ val staleConnection = NeverConnectedStateBleConnection()
+ val staleFactory =
+ object : BleConnectionFactory {
+ override fun create(scope: CoroutineScope, tag: String): BleConnection = staleConnection
+ }
+ val device = FakeBleDevice(address = address, name = "Test Radio")
+ bluetoothRepository.bond(device)
+ scanner.emitDevice(device)
+
+ var onDisconnectCalls = 0
+ every { service.onDisconnect(any(), any()) } calls { onDisconnectCalls++ }
+
+ val bleTransport =
+ BleRadioTransport(
+ scope = this,
+ scanner = scanner,
+ bluetoothRepository = bluetoothRepository,
+ connectionFactory = staleFactory,
+ callback = service,
+ address = address,
+ )
+ try {
+ bleTransport.start()
+
+ // Settle (3s) + CONNECTED_GATE_TIMEOUT (5s) + margin. connectAndAwait returns Connected,
+ // profile setup succeeds, but connectionState never emits Connected, so the gate times out.
+ advanceTimeBy(9_000L)
+
+ assertTrue(staleConnection.disconnectCalls >= 1, "Connected-gate timeout must force GATT disconnect")
+ assertTrue(onDisconnectCalls <= 1, "Connected-gate timeout must not spam onDisconnect callbacks")
+
+ val writesBefore = staleConnection.service.writes.size
+ bleTransport.handleSendToRadio(byteArrayOf(4, 5, 6))
+ assertEquals(
+ writesBefore,
+ staleConnection.service.writes.size,
+ "Connected-gate timeout must clear radioService so writes do not use stale profile state",
+ )
+
+ advanceTimeBy(10_000L)
+ assertTrue(
+ staleConnection.connectAndAwaitCalls >= 2,
+ "Reconnect loop must retry after Connected-gate timeout " +
+ "(actual calls: ${staleConnection.connectAndAwaitCalls})",
+ )
+
+ bleTransport.close()
+ } finally {
+ bleTransport.close()
+ }
+ }
+
+ /**
+ * Validates the normal Connected-gate path: when connectAndAwait returns Connected and the connectionState
+ * StateFlow reflects Connected, the gate succeeds immediately and the transport operates normally. The timeout path
+ * β where connectionState never reaches Connected β is covered separately by the `Connected-gate timeout cleans up
+ * stale service and retries without disconnect spam` test.
+ */
+ @Test
+ fun `Connected-gate normal path succeeds and transport operates`() = runTest {
+ val device = FakeBleDevice(address = address, name = "Test Radio")
+ bluetoothRepository.bond(device)
+ scanner.emitDevice(device)
+
+ val bleTransport =
+ BleRadioTransport(
+ scope = this,
+ scanner = scanner,
+ bluetoothRepository = bluetoothRepository,
+ connectionFactory = connectionFactory,
+ callback = service,
+ address = address,
+ )
+ every { service.onDisconnect(any(), any()) } returns Unit
+
+ try {
+ bleTransport.start()
+ // advance past connect β connectAndAwait returns Connected synchronously and the
+ // FakeBleConnection sets connectionState to Connected in connect(), so this test
+ // verifies the normal path where the gate succeeds.
+ advanceTimeBy(10_000L)
+ assertTrue(connection.connectAndAwaitCalls >= 1, "Must attempt at least one connection")
+
+ bleTransport.close()
+ } finally {
+ bleTransport.close()
+ }
}
}
@@ -329,3 +1069,52 @@ private class CancellingProfileBleConnection : BleConnection {
override fun maximumWriteValueLength(writeType: BleWriteType): Int? = null
}
+
+/**
+ * A [BleConnection] whose [connectAndAwait] reports success while [connectionState] never emits
+ * [BleConnectionState.Connected]. This exercises the Connected-gate timeout cleanup path after profile setup succeeds.
+ */
+private class NeverConnectedStateBleConnection : BleConnection {
+
+ private val _deviceFlow = MutableStateFlow<BleDevice?>(null)
+ override val deviceFlow: StateFlow<BleDevice?> = _deviceFlow.asStateFlow()
+
+ private val _connectionState = MutableStateFlow<BleConnectionState>(BleConnectionState.Disconnected())
+ override val connectionState: StateFlow<BleConnectionState> = _connectionState.asStateFlow()
+
+ override val device: BleDevice?
+ get() = _deviceFlow.value
+
+ val service = FakeBleService().apply { addCharacteristic(FROMNUM_CHARACTERISTIC) }
+
+ var connectAndAwaitCalls = 0
+ private set
+
+ var disconnectCalls = 0
+ private set
+
+ override suspend fun connect(device: BleDevice) {
+ _deviceFlow.value = device
+ _connectionState.value = BleConnectionState.Connecting
+ }
+
+ override suspend fun connectAndAwait(device: BleDevice, timeout: Duration): BleConnectionState {
+ connectAndAwaitCalls++
+ connect(device)
+ return BleConnectionState.Connected
+ }
+
+ override suspend fun disconnect() {
+ disconnectCalls++
+ _connectionState.value = BleConnectionState.Disconnected()
+ _deviceFlow.value = null
+ }
+
+ override suspend fun <T> profile(
+ serviceUuid: kotlin.uuid.Uuid,
+ timeout: Duration,
+ setup: suspend CoroutineScope.(BleService) -> T,
+ ): T = CoroutineScope(currentCoroutineContext()).setup(service)
+
+ override fun maximumWriteValueLength(writeType: BleWriteType): Int? = null
+}
diff --git a/core/network/src/commonTest/kotlin/org/meshtastic/core/network/radio/BleRadioTransportTest.kt b/core/network/src/commonTest/kotlin/org/meshtastic/core/network/radio/BleRadioTransportTest.kt
index 09e5ede0a8..8d88a2e7e1 100644
--- a/core/network/src/commonTest/kotlin/org/meshtastic/core/network/radio/BleRadioTransportTest.kt
+++ b/core/network/src/commonTest/kotlin/org/meshtastic/core/network/radio/BleRadioTransportTest.kt
@@ -72,11 +72,14 @@ class BleRadioTransportTest {
address = address,
)
bleTransport.start()
-
- // start() begins connect() which is async
- // In a real test we'd verify the connection state,
- // but for now this confirms it works with the fakes.
- assertEquals(address, bleTransport.address)
+ try {
+ // start() begins connect() which is async
+ // In a real test we'd verify the connection state,
+ // but for now this confirms it works with the fakes.
+ assertEquals(address, bleTransport.address)
+ } finally {
+ bleTransport.close()
+ }
}
@Test
@@ -106,7 +109,8 @@ class BleRadioTransportTest {
@Test
fun `onDisconnect is called after DEFAULT_FAILURE_THRESHOLD consecutive failures`() = runTest {
val device = FakeBleDevice(address = address, name = "Test Device")
- bluetoothRepository.bond(device) // skip BLE scan β device is already bonded
+ bluetoothRepository.bond(device)
+ scanner.emitDevice(device) // targeted scan resolves immediately; this test covers reconnect timing
// Make every connectAndAwait call throw so each iteration counts as one failure.
connection.connectException = RadioNotConnectedException("simulated failure")
@@ -122,15 +126,17 @@ class BleRadioTransportTest {
)
bleTransport.start()
- // Advance through exactly 3 failure iterations (β24 001 ms virtual time).
- // The 4th iteration's backoff hasn't elapsed yet, so the coroutine is suspended
- // and advanceTimeBy returns cleanly.
- advanceTimeBy(24_001L)
+ try {
+ // Advance through exactly 3 failure iterations (β24 001 ms virtual time).
+ // The 4th iteration's backoff hasn't elapsed yet, so the coroutine is suspended
+ // and advanceTimeBy returns cleanly.
+ advanceTimeBy(24_001L)
- verify { service.onDisconnect(any(), any()) }
-
- // Cancel the reconnect loop so runTest can complete.
- bleTransport.close()
+ verify { service.onDisconnect(any(), any()) }
+ } finally {
+ // Cancel the reconnect loop so runTest can complete.
+ bleTransport.close()
+ }
}
/**
@@ -147,6 +153,7 @@ class BleRadioTransportTest {
fun `reconnect loop never gives up - no permanent disconnect from policy`() = runTest {
val device = FakeBleDevice(address = address, name = "Test Device")
bluetoothRepository.bond(device)
+ scanner.emitDevice(device)
connection.connectException = RadioNotConnectedException("simulated failure")
every { service.onDisconnect(any(), any()) } returns Unit
@@ -162,15 +169,69 @@ class BleRadioTransportTest {
)
bleTransport.start()
- // Run well past where the legacy policy (maxFailures = 10) would have given up.
- advanceTimeBy(800_001L)
+ try {
+ // Run well past where the legacy policy (maxFailures = 10) would have given up.
+ advanceTimeBy(800_001L)
- // Transient disconnects (isPermanent = false) are expected once the failure threshold is hit;
- // the policy must NEVER signal a permanent disconnect on its own. Only explicit close()
- // (verified separately by the service layer) may emit isPermanent = true.
- verify(mode = VerifyMode.not) { service.onDisconnect(isPermanent = true, errorMessage = any()) }
+ // Transient disconnects (isPermanent = false) are expected once the failure threshold is hit;
+ // the policy must NEVER signal a permanent disconnect on its own. Only explicit close()
+ // (verified separately by the service layer) may emit isPermanent = true.
+ verify(mode = VerifyMode.not) { service.onDisconnect(isPermanent = true, errorMessage = any()) }
+ } finally {
+ bleTransport.close()
+ }
+ }
+
+ @Test
+ fun `findDevice prefers freshly scanned device over bonded device`() = runTest {
+ val bondedDevice = FakeBleDevice(address = address, name = "Bonded Device")
+ val scannedDevice = FakeBleDevice(address = address, name = "Scanned Device")
+ bluetoothRepository.bond(bondedDevice)
+ scanner.emitDevice(scannedDevice)
- bleTransport.close()
+ val bleTransport =
+ BleRadioTransport(
+ scope = this,
+ scanner = scanner,
+ bluetoothRepository = bluetoothRepository,
+ connectionFactory = connectionFactory,
+ callback = service,
+ address = address,
+ )
+ bleTransport.start()
+ try {
+ advanceTimeBy(3_001)
+
+ assertEquals("Scanned Device", connection.device?.name)
+ } finally {
+ bleTransport.close()
+ }
+ }
+
+ @Test
+ fun `findDevice falls back to bonded device when targeted scan finds nothing`() = runTest {
+ val bondedDevice = FakeBleDevice(address = address, name = "Bonded Device")
+ bluetoothRepository.bond(bondedDevice)
+
+ val bleTransport =
+ BleRadioTransport(
+ scope = this,
+ scanner = scanner,
+ bluetoothRepository = bluetoothRepository,
+ connectionFactory = connectionFactory,
+ callback = service,
+ address = address,
+ )
+ bleTransport.start()
+ try {
+ advanceTimeBy(5_500)
+
+ assertEquals(SERVICE_UUID, scanner.lastScanServiceUuid)
+ assertEquals(address, scanner.lastScanAddress)
+ assertEquals("Bonded Device", connection.device?.name)
+ } finally {
+ bleTransport.close()
+ }
}
@Test
@@ -188,12 +249,14 @@ class BleRadioTransportTest {
address = address,
)
bleTransport.start()
- advanceTimeBy(3_001)
-
- assertNotNull(scanner.lastScanServiceUuid, "scan must include serviceUuid")
- assertEquals(SERVICE_UUID, scanner.lastScanServiceUuid)
- assertEquals(address, scanner.lastScanAddress)
+ try {
+ advanceTimeBy(3_001)
- bleTransport.close()
+ assertNotNull(scanner.lastScanServiceUuid, "scan must include serviceUuid")
+ assertEquals(SERVICE_UUID, scanner.lastScanServiceUuid)
+ assertEquals(address, scanner.lastScanAddress)
+ } finally {
+ bleTransport.close()
+ }
}
}
diff --git a/core/service/build.gradle.kts b/core/service/build.gradle.kts
index 56153b853b..0a7b68f75f 100644
--- a/core/service/build.gradle.kts
+++ b/core/service/build.gradle.kts
@@ -59,6 +59,9 @@ kotlin {
}
}
- commonTest.dependencies { implementation(libs.kotlinx.coroutines.test) }
+ commonTest.dependencies {
+ implementation(projects.core.testing)
+ implementation(libs.kotlinx.coroutines.test)
+ }
}
}
diff --git a/core/service/src/commonMain/kotlin/org/meshtastic/core/service/SharedRadioInterfaceService.kt b/core/service/src/commonMain/kotlin/org/meshtastic/core/service/SharedRadioInterfaceService.kt
index 68a3573e9c..09a083f532 100644
--- a/core/service/src/commonMain/kotlin/org/meshtastic/core/service/SharedRadioInterfaceService.kt
+++ b/core/service/src/commonMain/kotlin/org/meshtastic/core/service/SharedRadioInterfaceService.kt
@@ -128,12 +128,28 @@ class SharedRadioInterfaceService(
*/
@Volatile private var isStopping = false
+ /** Prevents concurrent liveness-induced transport restarts from stacking. */
+ private val isRestarting = atomic(false)
+
private val listenersInitialized = atomic(false)
private var heartbeatJob: Job? = null
private var lastHeartbeatMillis = 0L
@Volatile private var lastDataReceivedMillis = 0L
+ /**
+ * Internal test seam for deterministic clock injection. Production uses [nowMillis]; tests override this to a
+ * controllable clock so [onConnect], [handleFromRadio], [checkLiveness], and [keepAlive] all share one coherent
+ * time source. Not a constructor parameter to avoid breaking Koin @Single annotation generation (which would try to
+ * resolve `() -> Long` from the DI graph).
+ */
+ @Volatile
+ @Suppress("MemberVisibilityCanBePrivate")
+ internal var clockMillis: () -> Long = { nowMillis }
+
+ /** The current time from the injected clock. */
+ private fun now(): Long = clockMillis()
+
companion object {
private const val HEARTBEAT_INTERVAL_MILLIS = 30 * 1000L
@@ -271,8 +287,16 @@ class SharedRadioInterfaceService(
startHeartbeat()
}
- /** Must be called under [transportMutex]. */
- private suspend fun stopTransportLocked() {
+ /**
+ * Must be called under [transportMutex].
+ *
+ * @param notifyPermanent When `true`, emits a permanent disconnect state to [connectionState]. Set `false` during
+ * automatic liveness recovery to avoid surfacing a user-facing disconnect.
+ * @param sendPoliteDisconnect When `true`, sends a `ToRadio(disconnect = true)` frame to the firmware before
+ * tearing down. Set `false` when the transport is already dead (zombie session) to avoid writing into a broken
+ * link.
+ */
+ private suspend fun stopTransportLocked(notifyPermanent: Boolean = true, sendPoliteDisconnect: Boolean = true) {
val currentTransport = radioTransport
Logger.i { "Stopping transport $currentTransport" }
// Best-effort polite goodbye: tell the firmware we're disconnecting on purpose so it can
@@ -284,7 +308,9 @@ class SharedRadioInterfaceService(
// transport's own scope; the drain delay gives async transports a window to flush before
// close() cancels their write scope. BLE's retry path backs off 500ms, so this window
// also covers one retry on flaky GATT links.
- if (currentTransport != null && _connectionState.value != ConnectionState.Disconnected) {
+ if (
+ sendPoliteDisconnect && currentTransport != null && _connectionState.value != ConnectionState.Disconnected
+ ) {
isStopping = true
ignoreExceptionSuspend {
currentTransport.handleSendToRadio(ToRadio(disconnect = true).encode())
@@ -300,14 +326,14 @@ class SharedRadioInterfaceService(
_serviceScope.cancel("stopping transport")
_serviceScope = CoroutineScope(dispatchers.io + SupervisorJob())
- if (currentTransport != null) {
+ if (notifyPermanent && currentTransport != null) {
onDisconnect(isPermanent = true)
}
}
private fun startHeartbeat() {
heartbeatJob?.cancel()
- lastDataReceivedMillis = nowMillis
+ lastDataReceivedMillis = now()
heartbeatJob =
serviceScope.launch {
while (true) {
@@ -323,21 +349,60 @@ class SharedRadioInterfaceService(
*
* If we believe we're connected but haven't received any data from the radio within [LIVENESS_TIMEOUT_MILLIS], the
* connection is likely dead. Signal a non-permanent disconnect so the reconnect machinery can take over.
+ *
+ * Uses [clockMillis] for the current time so tests can inject a deterministic clock.
*/
- private fun checkLiveness() {
+ internal fun checkLiveness() {
if (_connectionState.value != ConnectionState.Connected) return
- val silenceMs = nowMillis - lastDataReceivedMillis
+ val silenceMs = now() - lastDataReceivedMillis
if (silenceMs > LIVENESS_TIMEOUT_MILLIS) {
+ // "Silence" = lastDataReceivedMillis not updated by handleFromRadio (no inbound
+ // packets). Only BLE suffers from silent zombie sessions (no disconnect signal from
+ // stack), so the liveness-restart path is BLE-only. For non-BLE transports we return
+ // WITHOUT emitting a disconnect or mutating ConnectionState β there is no
+ // transport-level timeout contract proving that silence past this threshold means
+ // the session is dead.
+ if (runningTransportId != InterfaceId.BLUETOOTH) {
+ Logger.d { "Ignoring liveness timeout for non-BLE transport (silence: ${silenceMs}ms)" }
+ return
+ }
+
Logger.w {
"Liveness check failed: no data received for ${silenceMs}ms " +
- "(threshold: ${LIVENESS_TIMEOUT_MILLIS}ms). Treating as disconnect."
+ "(threshold: ${LIVENESS_TIMEOUT_MILLIS}ms). Restarting BLE transport."
+ }
+
+ // Force transport restart to recover from silent zombie sessions where the BLE stack
+ // did not report a disconnect. Uses the same processLifecycle scope and transportMutex
+ // pattern as setDeviceAddress() to guarantee clean teardown/restart sequencing.
+ // The onDisconnect notification is emitted INSIDE the compareAndSet guard so that a
+ // double liveness-timeout (timer not cancelled between fires) does not produce
+ // duplicate disconnect notifications for a single restart cycle.
+ if (isRestarting.compareAndSet(expect = false, update = true)) {
+ // Silent recovery: emit the non-permanent state transition (DeviceSleep) so the
+ // reconnect machinery takes over, but do NOT pass an errorMessage. Automatic
+ // liveness recovery is self-healing β surfacing a modal dialog for a transient
+ // condition the app already handled is confusing UX. The warning log above
+ // remains the observability surface for this event.
+ onDisconnect(isPermanent = false)
+ processLifecycle.coroutineScope.launch {
+ try {
+ transportMutex.withLock {
+ ignoreExceptionSuspend {
+ stopTransportLocked(notifyPermanent = false, sendPoliteDisconnect = false)
+ }
+ startTransportLocked()
+ }
+ } finally {
+ isRestarting.value = false
+ }
+ }
}
- onDisconnect(isPermanent = false, errorMessage = "Connection timeout β no data received")
}
}
- fun keepAlive(now: Long = nowMillis) {
+ fun keepAlive(now: Long = now()) {
if (now - lastHeartbeatMillis > HEARTBEAT_INTERVAL_MILLIS) {
radioTransport?.keepAlive()
lastHeartbeatMillis = now
@@ -369,7 +434,7 @@ class SharedRadioInterfaceService(
@Suppress("TooGenericExceptionCaught")
override fun handleFromRadio(bytes: ByteArray) {
try {
- lastDataReceivedMillis = nowMillis
+ lastDataReceivedMillis = now()
// trySend synchronously onto the unbounded Channel so packet order matches arrival
// order. The previous `launch { emit() }` pattern dispatched each packet onto a
// fresh coroutine, letting the scheduler reorder them β which broke the firmware
@@ -397,7 +462,7 @@ class SharedRadioInterfaceService(
// launching a coroutine. The async launch pattern introduced a window where a concurrent
// onDisconnect launch could execute AFTER an onConnect launch, leaving the service stuck
// in Connected while the transport was actually disconnected.
- lastDataReceivedMillis = nowMillis
+ lastDataReceivedMillis = now()
if (_connectionState.value != ConnectionState.Connected) {
Logger.d { "Broadcasting connection state change to Connected" }
_connectionState.value = ConnectionState.Connected
diff --git a/core/service/src/commonTest/kotlin/org/meshtastic/core/service/SharedRadioInterfaceServiceLivenessTest.kt b/core/service/src/commonTest/kotlin/org/meshtastic/core/service/SharedRadioInterfaceServiceLivenessTest.kt
new file mode 100644
index 0000000000..227898c08b
--- /dev/null
+++ b/core/service/src/commonTest/kotlin/org/meshtastic/core/service/SharedRadioInterfaceServiceLivenessTest.kt
@@ -0,0 +1,553 @@
+/*
+ * Copyright (c) 2026 Meshtastic LLC
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program. If not, see <https://www.gnu.org/licenses/>.
+ */
+package org.meshtastic.core.service
+
+import androidx.lifecycle.Lifecycle
+import androidx.lifecycle.LifecycleEventObserver
+import androidx.lifecycle.LifecycleObserver
+import androidx.lifecycle.LifecycleOwner
+import dev.mokkery.MockMode
+import dev.mokkery.answering.calls
+import dev.mokkery.answering.returns
+import dev.mokkery.every
+import dev.mokkery.matcher.any
+import dev.mokkery.mock
+import kotlinx.coroutines.CompletableDeferred
+import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.ExperimentalCoroutinesApi
+import kotlinx.coroutines.flow.MutableSharedFlow
+import kotlinx.coroutines.flow.MutableStateFlow
+import kotlinx.coroutines.launch
+import kotlinx.coroutines.test.UnconfinedTestDispatcher
+import kotlinx.coroutines.test.advanceTimeBy
+import kotlinx.coroutines.test.resetMain
+import kotlinx.coroutines.test.runTest
+import kotlinx.coroutines.test.setMain
+import org.meshtastic.core.di.CoroutineDispatchers
+import org.meshtastic.core.model.ConnectionState
+import org.meshtastic.core.model.DeviceType
+import org.meshtastic.core.network.repository.NetworkRepository
+import org.meshtastic.core.repository.PlatformAnalytics
+import org.meshtastic.core.repository.RadioTransport
+import org.meshtastic.core.repository.RadioTransportFactory
+import org.meshtastic.core.testing.FakeBluetoothRepository
+import org.meshtastic.core.testing.FakeRadioPrefs
+import org.meshtastic.core.testing.FakeRadioTransport
+import kotlin.test.AfterTest
+import kotlin.test.BeforeTest
+import kotlin.test.Test
+import kotlin.test.assertEquals
+import kotlin.test.assertFalse
+import kotlin.test.assertTrue
+
+/**
+ * Service-level tests for [SharedRadioInterfaceService] liveness detection.
+ *
+ * Uses a controllable clock via [SharedRadioInterfaceService.clockMillis] so [onConnect], [handleFromRadio], and
+ * [checkLiveness] all share one coherent time source β no mixing of real wall-clock with test time.
+ *
+ * A counting transport factory returns a fresh [FakeRadioTransport] per createTransport() call so we can observe how
+ * many restarts actually occurred.
+ */
+@OptIn(ExperimentalCoroutinesApi::class)
+class SharedRadioInterfaceServiceLivenessTest {
+
+ private val testDispatcher = UnconfinedTestDispatcher()
+ private val dispatchers = CoroutineDispatchers(io = testDispatcher, main = testDispatcher, default = testDispatcher)
+
+ private lateinit var processLifecycleOwner: TestLifecycleOwner
+
+ @BeforeTest
+ fun setUp() {
+ // processLifecycle.coroutineScope uses Dispatchers.Main.immediate internally;
+ // JVM tests must install a Main dispatcher or get IllegalStateException.
+ Dispatchers.setMain(testDispatcher)
+ // Create the lifecycle owner AFTER setMain so Robolectric's main thread is ready.
+ // Field initializers run before @BeforeTest, which is too early for Robolectric.
+ processLifecycleOwner = TestLifecycleOwner()
+ }
+
+ @AfterTest
+ fun tearDown() {
+ // Release any suspended close gate so a held in-flight restart can complete; otherwise
+ // disconnect() below would block forever on the gated transport's close().
+ // NOTE: relies on UnconfinedTestDispatcher resuming the gated close() inline when
+ // complete(Unit) is called β if testDispatcher is ever changed to StandardTestDispatcher,
+ // add testDispatcher.scheduler.runCurrent() here before runBlocking to avoid a mutex deadlock.
+ activeCloseGate?.complete(Unit)
+ activeCloseGate = null
+ // Service cleanup is handled per-test in try/finally blocks β each test calls
+ // service.disconnect() + advanceTimeBy in a finally clause. tearDown cannot use
+ // runBlocking { services.forEach { it.disconnect() } } because it deadlocks on
+ // Robolectric's main thread (androidHostTest target).
+ services.clear()
+ createdTransports.clear()
+ // CRITICAL: Destroy the lifecycle to cancel processLifecycle.coroutineScope and all
+ // leaked collectors (devAddr, bluetoothRepository.state, networkRepository.networkAvailable).
+ // Without this, those infinite flow collectors keep the forked test JVM alive after tests
+ // complete, causing Gradle to hang at subsequent :core:*:allTests tasks.
+ processLifecycleOwner.destroy()
+ // Let pending cancellations propagate before resetting the Main dispatcher. Use runCurrent
+ // (NOT advanceUntilIdle): the test bodies already disconnect every service, so no heartbeat
+ // loop should be active here β advanceUntilIdle would hang if one somehow survived.
+ testDispatcher.scheduler.runCurrent()
+ Dispatchers.resetMain()
+ }
+
+ private val bluetoothRepository = FakeBluetoothRepository()
+ private val radioPrefs = FakeRadioPrefs()
+
+ private val networkRepository: NetworkRepository = mock(MockMode.autofill)
+ private val analytics: PlatformAnalytics = mock(MockMode.autofill)
+
+ /**
+ * Minimal [LifecycleOwner] for tests that avoids [LifecycleRegistry], which enforces main-thread checks and throws
+ * `RuntimeException` under Robolectric (androidHostTest). This custom [Lifecycle] dispatches ON_DESTROY to
+ * registered [LifecycleEventObserver]s so `lifecycleScope` cancels correctly.
+ */
+ private class TestLifecycleOwner : LifecycleOwner {
+ private val observers = mutableListOf<LifecycleObserver>()
+ private var state = Lifecycle.State.RESUMED
+
+ override val lifecycle: Lifecycle =
+ object : Lifecycle() {
+ override fun addObserver(observer: LifecycleObserver) {
+ observers.add(observer)
+ }
+
+ override fun removeObserver(observer: LifecycleObserver) {
+ observers.remove(observer)
+ }
+
+ override val currentState: Lifecycle.State
+ get() = state
+ }
+
+ fun destroy() {
+ state = Lifecycle.State.DESTROYED
+ val event = Lifecycle.Event.ON_DESTROY
+ observers.toList().forEach { observer ->
+ (observer as? LifecycleEventObserver)?.onStateChanged(this@TestLifecycleOwner, event)
+ }
+ }
+ }
+
+ /**
+ * Test-only [RadioTransport] whose [close] suspends on a [CompletableDeferred] gate.
+ *
+ * The liveness restart path calls `stopTransportLocked` β `currentTransport.close()` inside a launched coroutine.
+ * With the default [FakeRadioTransport], `close()` returns without suspending, so under [UnconfinedTestDispatcher]
+ * the entire restart completes synchronously during `checkLiveness()` and a second `checkLiveness()` never observes
+ * an in-flight restart. By awaiting a gate inside `close()`, this fake holds the restart genuinely suspended
+ * mid-flight, letting a test deterministically exercise the in-flight overlap window and prove the second check
+ * does not stack another restart/close.
+ *
+ * The gate is shared across instances; once completed by the test, any pending or subsequent `close()` resumes
+ * immediately.
+ */
+ private class GatedFakeRadioTransport(private val closeGate: CompletableDeferred<Unit>) : RadioTransport {
+ var closeCalled = false
+ private set
+
+ var closeCount = 0
+ private set
+
+ var closeCompletedCount = 0
+ private set
+
+ // Liveness restart skips the polite-disconnect frame (sendPoliteDisconnect = false), so no
+ // outbound data is expected; satisfy the contract with a no-op.
+ override fun handleSendToRadio(p: ByteArray) = Unit
+
+ override suspend fun close() {
+ closeCalled = true
+ closeCount++
+ // Suspend here until the test releases the gate, holding the restart in-flight.
+ closeGate.await()
+ closeCompletedCount++
+ }
+ }
+
+ /** Controllable clock β tests advance this manually so all time comparisons are deterministic. */
+ private var clock: Long = 0L
+
+ /**
+ * Tracks every [SharedRadioInterfaceService] created via [createConnectedService] so [tearDown] can disconnect them
+ * deterministically. Destroying the process lifecycle does NOT cancel the service's private `_serviceScope` (which
+ * hosts the heartbeat loop), so we must call `disconnect()` explicitly.
+ */
+ private val services = mutableListOf<SharedRadioInterfaceService>()
+
+ /**
+ * Tracks the suspended close gate for the in-flight restart test so [tearDown] can release it even if the test body
+ * throws. Without this, a failed assertion before `closeGate.complete(Unit)` would leave a restart suspended
+ * forever and hang teardown.
+ */
+ private var activeCloseGate: CompletableDeferred<Unit>? = null
+
+ /** Tracks all transports created by the factory so we can count restarts and inspect sent data. */
+ private val createdTransports = mutableListOf<FakeRadioTransport>()
+ private val transportFactory: RadioTransportFactory = mock(MockMode.autofill)
+
+ /**
+ * Creates a [SharedRadioInterfaceService] with a controllable clock and a factory that returns a fresh
+ * [FakeRadioTransport] per createTransport() call. After construction, calls [connect] then explicitly [onConnect]
+ * to bring the service to Connected state (FakeRadioTransport does not call onConnect itself).
+ *
+ * Pass [transportProvider] to swap in a custom test double (e.g. a suspending-close fake) instead of the default
+ * [FakeRadioTransport]; the default records each created transport in [createdTransports].
+ */
+ private fun createConnectedService(
+ address: String,
+ transportProvider: () -> RadioTransport = { FakeRadioTransport().also { createdTransports.add(it) } },
+ ): SharedRadioInterfaceService {
+ every { networkRepository.networkAvailable } returns MutableStateFlow(true)
+ every { networkRepository.resolvedList } returns MutableSharedFlow()
+ every { analytics.isPlatformServicesAvailable } returns false
+ every { transportFactory.supportedDeviceTypes } returns listOf(DeviceType.BLE)
+ every { transportFactory.isMockTransport() } returns false
+ every { transportFactory.isAddressValid(any()) } returns true
+ every { transportFactory.toInterfaceAddress(any(), any()) } returns address
+ every { transportFactory.createTransport(any(), any()) } calls { transportProvider() }
+
+ radioPrefs.setDevAddr(address)
+
+ val service =
+ SharedRadioInterfaceService(
+ dispatchers = dispatchers,
+ bluetoothRepository = bluetoothRepository,
+ networkRepository = networkRepository,
+ processLifecycle = processLifecycleOwner.lifecycle,
+ radioPrefs = radioPrefs,
+ transportFactory = transportFactory,
+ analytics = analytics,
+ )
+ service.clockMillis = { clock }
+ // Register the service so tearDown can disconnect it deterministically (the heartbeat loop
+ // launched in _serviceScope would otherwise outlive the test).
+ services.add(service)
+ service.connect()
+ service.onConnect()
+ return service
+ }
+
+ // βββ BLE: Liveness timeout triggers recovery βββββββββββββββββββββββββββββββββββββββββββββββ
+
+ @Test
+ fun `BLE liveness timeout closes old transport and creates fresh one`() = runTest(testDispatcher) {
+ clock = 0L
+ val service = createConnectedService("xAA:BB:CC:DD:EE:FF")
+ try {
+ assertEquals(1, createdTransports.size, "Initial connect should create one transport")
+
+ clock = 65_000L
+ service.checkLiveness()
+ // Under UnconfinedTestDispatcher the liveness restart (sendPoliteDisconnect = false) runs
+ // inline during checkLiveness(). runCurrent/advanceTimeBy are belt-and-suspenders; the
+ // real 500ms polite-disconnect delay is covered by the trailing service.disconnect() below.
+ testDispatcher.scheduler.runCurrent()
+ advanceTimeBy(1_000L)
+
+ assertEquals(2, createdTransports.size, "Liveness restart should create exactly one fresh transport")
+ assertTrue(createdTransports.first().closeCalled, "Old transport must be closed")
+ assertEquals(1, createdTransports.first().closeCount, "Old transport closed exactly once")
+ } finally {
+ service.disconnect()
+ advanceTimeBy(1_000L)
+ }
+ }
+
+ @Test
+ fun `BLE liveness restart does not emit permanent Disconnected`() = runTest(testDispatcher) {
+ clock = 0L
+ val service = createConnectedService("xAA:BB:CC:DD:EE:FF")
+
+ try {
+ // Capture all state transitions during the liveness recovery
+ val stateEmissions = mutableListOf<ConnectionState>()
+ val collectJob = backgroundScope.launch { service.connectionState.collect { stateEmissions.add(it) } }
+
+ clock = 65_000L
+ service.checkLiveness()
+ // The restart completes inline under UnconfinedTestDispatcher; runCurrent/advanceTimeBy
+ // are belt-and-suspenders. The trailing disconnect() covers its own 500ms polite delay.
+ testDispatcher.scheduler.runCurrent()
+ advanceTimeBy(1_000L)
+
+ collectJob.cancel()
+
+ // Recovery must NEVER emit permanent Disconnected
+ assertFalse(
+ ConnectionState.Disconnected in stateEmissions,
+ "Automatic recovery must not emit permanent Disconnected state " + "(emitted: $stateEmissions)",
+ )
+ } finally {
+ service.disconnect()
+ advanceTimeBy(1_000L)
+ }
+ }
+
+ @Test
+ fun `BLE liveness restart does not emit user-facing connection error`() = runTest(testDispatcher) {
+ clock = 0L
+ val service = createConnectedService("xAA:BB:CC:DD:EE:FF")
+
+ try {
+ // Collect connectionError emissions β automatic liveness recovery must be silent.
+ // _connectionError is a no-replay SharedFlow, so the collector must subscribe before
+ // triggering the liveness timeout. Under UnconfinedTestDispatcher the launch runs
+ // eagerly to its first suspension (awaiting SharedFlow emission).
+ val errors = mutableListOf<String>()
+ val collectJob = backgroundScope.launch { service.connectionError.collect { errors.add(it) } }
+
+ clock = 65_000L
+ service.checkLiveness()
+ // The restart completes inline under UnconfinedTestDispatcher; runCurrent/advanceTimeBy
+ // are belt-and-suspenders (mirrors the sibling liveness tests' pattern).
+ testDispatcher.scheduler.runCurrent()
+ advanceTimeBy(1_000L)
+
+ collectJob.cancel()
+
+ assertTrue(
+ errors.isEmpty(),
+ "Automatic BLE liveness recovery must not emit user-facing connection error (got: $errors)",
+ )
+ } finally {
+ service.disconnect()
+ advanceTimeBy(1_000L)
+ }
+ }
+
+ @Test
+ fun `BLE liveness restart does not send polite disconnect into zombie transport`() = runTest(testDispatcher) {
+ clock = 0L
+ val service = createConnectedService("xAA:BB:CC:DD:EE:FF")
+ try {
+ val oldTransport = createdTransports.first()
+
+ oldTransport.sentData.clear()
+
+ clock = 65_000L
+ service.checkLiveness()
+ testDispatcher.scheduler.runCurrent()
+ advanceTimeBy(1_000L)
+
+ assertTrue(
+ oldTransport.sentData.isEmpty(),
+ "Polite disconnect frame must NOT be sent into zombie transport during liveness restart",
+ )
+ } finally {
+ service.disconnect()
+ advanceTimeBy(1_000L)
+ }
+ }
+
+ @Test
+ fun `BLE repeated liveness checks do not stack restarts`() = runTest(testDispatcher) {
+ clock = 0L
+ val service = createConnectedService("xAA:BB:CC:DD:EE:FF")
+
+ try {
+ clock = 65_000L
+ service.checkLiveness()
+ testDispatcher.scheduler.runCurrent()
+ advanceTimeBy(1_000L)
+
+ clock = 66_000L
+ service.checkLiveness()
+ testDispatcher.scheduler.runCurrent()
+ advanceTimeBy(1_000L)
+
+ val firstTransportCloses = createdTransports.firstOrNull()?.closeCount ?: 0
+ assertEquals(1, firstTransportCloses, "First transport should be closed exactly once (no stacking)")
+ } finally {
+ service.disconnect()
+ advanceTimeBy(1_000L)
+ }
+ }
+
+ @Test
+ fun `BLE in-flight liveness restart prevents overlapping restart via isRestarting`() = runTest(testDispatcher) {
+ // Deterministic in-flight overlap: a GatedFakeRadioTransport holds the first restart
+ // genuinely suspended inside stopTransportLocked β close() (awaiting closeGate). This
+ // removes reliance on UnconfinedTestDispatcher scheduling so the overlap window is real.
+ //
+ // The first checkLiveness() flips state to DeviceSleep and CAS-sets isRestarting before
+ // launching the restart coroutine, which then suspends in close(). The second
+ // checkLiveness() is issued while that restart is still suspended and must NOT begin
+ // another close/create cycle.
+ val gatedTransports = mutableListOf<GatedFakeRadioTransport>()
+ val closeGate = CompletableDeferred<Unit>()
+ // Publish the gate to activeCloseGate so tearDown can release it even if an assertion below
+ // throws before we reach the try/finally β otherwise disconnect() would hang on close().
+ activeCloseGate = closeGate
+ val transportProvider: () -> RadioTransport = {
+ GatedFakeRadioTransport(closeGate).also { gatedTransports.add(it) }
+ }
+
+ clock = 0L
+ val service = createConnectedService("xAA:BB:CC:DD:EE:FF", transportProvider)
+ try {
+ assertEquals(1, gatedTransports.size, "Initial connect should create one transport")
+ val initialTransport = gatedTransports.first()
+
+ // Past the 60s threshold β first checkLiveness triggers a restart whose close() suspends
+ // on closeGate. Under UnconfinedTestDispatcher the launched restart runs eagerly up to the
+ // suspension point, so by the time checkLiveness() returns the restart is in-flight.
+ clock = 65_000L
+ service.checkLiveness()
+
+ // Issue a second checkLiveness() while the first restart is still suspended in close().
+ // Do NOT advance time here β the overlap must happen with the first restart in-flight.
+ clock = 65_001L
+ service.checkLiveness()
+
+ // Assertions below run while the restart is held suspended on closeGate. They MUST be
+ // wrapped in try/finally so closeGate is completed even if one of them fails; otherwise
+ // tearDown's runBlocking { disconnect() } would hang forever on the gated close().
+ try {
+ // While the first restart is suspended: exactly one transport created so far, and close()
+ // was entered exactly once and has NOT completed. The second check started no new cycle.
+ assertEquals(
+ 1,
+ gatedTransports.size,
+ "Second check must not create a transport while the first restart is in-flight",
+ )
+ assertTrue(
+ initialTransport.closeCalled,
+ "First transport close must have been entered by the restart",
+ )
+ assertEquals(
+ 1,
+ initialTransport.closeCount,
+ "close() entered exactly once (no stacking of close calls)",
+ )
+ assertEquals(
+ 0,
+ initialTransport.closeCompletedCount,
+ "close() must still be suspended (restart held in-flight) before releasing the gate",
+ )
+ } finally {
+ // Release the gate unconditionally so the suspended restart can complete. tearDown
+ // also releases activeCloseGate, but completing it here is required for the post-finally
+ // assertions below to observe the resumed restart.
+ closeGate.complete(Unit)
+ }
+
+ // Release the gate: the suspended restart resumes, completes stopTransportLocked (whose
+ // polite-disconnect delay is 500ms β covered by the 1s below), and startTransportLocked
+ // creates the single fresh transport. isRestarting is reset in the finally block.
+ testDispatcher.scheduler.runCurrent()
+ advanceTimeBy(1_000L)
+
+ // Exactly 2 transports: 1 initial + 1 restart. A stacking bug would produce 3+.
+ assertEquals(
+ 2,
+ gatedTransports.size,
+ "Exactly one fresh transport created after the restart resumes (1 initial + 1 restart)",
+ )
+ assertEquals(
+ 1,
+ initialTransport.closeCount,
+ "First transport still closed exactly once after restart completes",
+ )
+ assertEquals(1, initialTransport.closeCompletedCount, "First transport close completed exactly once")
+ } finally {
+ service.disconnect()
+ advanceTimeBy(1_000L)
+ }
+ }
+
+ // βββ Non-BLE: Liveness does not mutate state βββββββββββββββββββββββββββββββββββββββββββββββ
+
+ @Test
+ fun `non-BLE transport liveness timeout does not close transport or change state`() = runTest(testDispatcher) {
+ clock = 0L
+ val service = createConnectedService("t192.168.1.100")
+ try {
+ val stateBefore = service.connectionState.value
+
+ clock = 65_000L
+ service.checkLiveness()
+ testDispatcher.scheduler.runCurrent()
+ advanceTimeBy(1_000L)
+
+ assertEquals(stateBefore, service.connectionState.value, "Non-BLE state must not change")
+ assertFalse(createdTransports.first().closeCalled, "Non-BLE transport must NOT be closed")
+ assertEquals(1, createdTransports.size, "No restart should occur for non-BLE transport")
+ } finally {
+ service.disconnect()
+ advanceTimeBy(1_000L)
+ }
+ }
+
+ // βββ handleFromRadio resets the liveness timer ββββββββββββββββββββββββββββββββββββββββββββββ
+
+ @Test
+ fun `inbound data resets liveness timer so timeout does not fire`() = runTest(testDispatcher) {
+ clock = 0L
+ val service = createConnectedService("xAA:BB:CC:DD:EE:FF")
+
+ try {
+ // Advance 30s, then receive data (resets lastDataReceivedMillis to clock=30s)
+ clock = 30_000L
+ service.handleFromRadio(byteArrayOf(1, 2, 3))
+
+ // 30s since last data β within 60s threshold β should NOT fire
+ clock = 60_000L
+ service.checkLiveness()
+ assertFalse(
+ createdTransports.first().closeCalled,
+ "Liveness must not fire when silence is within threshold after inbound data",
+ )
+
+ // 66s since last data (at t=30s) β past 60s threshold β should fire
+ clock = 96_000L
+ service.checkLiveness()
+ testDispatcher.scheduler.runCurrent()
+ advanceTimeBy(1_000L)
+ assertTrue(
+ createdTransports.first().closeCalled,
+ "Liveness should fire after silence exceeds threshold since last inbound data",
+ )
+ } finally {
+ service.disconnect()
+ advanceTimeBy(1_000L)
+ }
+ }
+
+ @Test
+ fun `BLE liveness does not fire when connection state is not Connected`() = runTest(testDispatcher) {
+ clock = 0L
+ val service = createConnectedService("xAA:BB:CC:DD:EE:FF")
+
+ try {
+ service.onDisconnect(isPermanent = true)
+ assertFalse(service.connectionState.value == ConnectionState.Connected)
+
+ clock = 65_000L
+ service.checkLiveness()
+ testDispatcher.scheduler.runCurrent()
+ advanceTimeBy(1_000L)
+ assertFalse(createdTransports.first().closeCalled, "Liveness must not fire when not Connected")
+ } finally {
+ service.disconnect()
+ advanceTimeBy(1_000L)
+ }
+ }
+}
diff --git a/core/testing/src/commonMain/kotlin/org/meshtastic/core/testing/FakeBle.kt b/core/testing/src/commonMain/kotlin/org/meshtastic/core/testing/FakeBle.kt
index 331ac60ebc..efaf16a220 100644
--- a/core/testing/src/commonMain/kotlin/org/meshtastic/core/testing/FakeBle.kt
+++ b/core/testing/src/commonMain/kotlin/org/meshtastic/core/testing/FakeBle.kt
@@ -24,6 +24,7 @@ import kotlinx.coroutines.flow.StateFlow
import kotlinx.coroutines.flow.asStateFlow
import kotlinx.coroutines.flow.emitAll
import kotlinx.coroutines.flow.flow
+import kotlinx.coroutines.flow.onStart
import org.meshtastic.core.ble.BleCharacteristic
import org.meshtastic.core.ble.BleConnection
import org.meshtastic.core.ble.BleConnectionFactory
@@ -170,6 +171,9 @@ class FakeBleConnection :
if (serviceUuid in missingServices) {
throw NoSuchElementException("Service $serviceUuid not found")
}
+ // Use Dispatchers.Unconfined so notification emissions are delivered synchronously to
+ // collectors (write β immediate notification). This matches the original FakeBleConnection
+ // contract and the auto-responding pattern used by DFU/OTA transport tests.
return CoroutineScope(Dispatchers.Unconfined).setup(service)
}
@@ -193,18 +197,89 @@ class FakeBleService : BleService {
val writes = mutableListOf<FakeBleWrite>()
+ /** When non-null, [write] throws this exception on every call until explicitly cleared. */
+ var writeException: Exception? = null
+
+ /**
+ * When non-null, [read] throws this exception instead of returning data. Reset to null before throwing (in the same
+ * call).
+ */
+ var readException: Exception? = null
+
+ /**
+ * When non-null, [observe] returns a flow that immediately throws. Reset to null when observe() is called (before
+ * flow collection).
+ */
+ var observeException: Exception? = null
+
+ /** Characteristic-specific observe failures that occur when the returned flow is collected. */
+ val observeExceptionsByCharacteristic: MutableMap<Uuid, Exception> = mutableMapOf()
+
+ /** Characteristic-specific observe failures that occur before [BleService.observe]'s onSubscription callback. */
+ val observeBeforeSubscriptionExceptionByCharacteristic: MutableMap<Uuid, Exception> = mutableMapOf()
+
+ /** Characteristics whose 2-arg [observe] never invokes onSubscription. Notifications can still be emitted. */
+ val observeNeverSubscribeCharacteristics: MutableSet<Uuid> = mutableSetOf()
+
override fun hasCharacteristic(characteristic: BleCharacteristic): Boolean =
availableCharacteristics.contains(characteristic.uuid)
- override fun observe(characteristic: BleCharacteristic): Flow<ByteArray> =
- notificationFlows.getOrPut(characteristic.uuid) { MutableSharedFlow(extraBufferCapacity = 16) }
+ override fun observe(characteristic: BleCharacteristic): Flow<ByteArray> {
+ val failure =
+ observeExceptionsByCharacteristic.remove(characteristic.uuid)
+ ?: observeException?.also { observeException = null }
+ if (failure != null) {
+ return flow { throw failure }
+ }
+ return notificationFlows.getOrPut(characteristic.uuid) { MutableSharedFlow(extraBufferCapacity = 16) }
+ }
+
+ /**
+ * Overrides the 2-arg observe to prevent false subscriptionReady when testing pre-readiness failures.
+ *
+ * The default BleService implementation calls `observe(characteristic).onStart { onSubscription() }`, which would
+ * invoke [onSubscription] BEFORE any pre-collection failure flow throws. This override throws BEFORE invoking
+ * [onSubscription], correctly simulating "observe failed before CCCD/subscription readiness."
+ *
+ * Pre-readiness failures are sourced (in priority order) from:
+ * - [observeBeforeSubscriptionExceptionByCharacteristic] (2-arg-specific, one-shot per uuid)
+ * - [observeExceptionsByCharacteristic] (shared with 1-arg observe, one-shot per uuid; defensively treated as a
+ * pre-subscription failure here so the bare onStart wrap cannot swallow it after [onSubscription] runs)
+ * - [observeException] (global, one-shot)
+ *
+ * For characteristics in [observeNeverSubscribeCharacteristics], [onSubscription] is never invoked but
+ * notifications are still exposed β the returned flow is the bare SharedFlow with no [onStart] wrap, so
+ * [emitNotification] still reaches active collectors.
+ */
+ override fun observe(characteristic: BleCharacteristic, onSubscription: suspend () -> Unit): Flow<ByteArray> {
+ val failure =
+ observeBeforeSubscriptionExceptionByCharacteristic.remove(characteristic.uuid)
+ ?: observeExceptionsByCharacteristic.remove(characteristic.uuid)
+ ?: observeException?.also { observeException = null }
+ if (failure != null) {
+ // onSubscription is NOT invoked β simulates failure before CCCD/subscription readiness.
+ return flow { throw failure }
+ }
+ val base = observe(characteristic)
+ return if (characteristic.uuid in observeNeverSubscribeCharacteristics) {
+ base
+ } else {
+ base.onStart { onSubscription() }
+ }
+ }
- override suspend fun read(characteristic: BleCharacteristic): ByteArray =
- readQueues[characteristic.uuid]?.removeFirstOrNull() ?: ByteArray(0)
+ override suspend fun read(characteristic: BleCharacteristic): ByteArray {
+ readException?.let {
+ readException = null
+ throw it
+ }
+ return readQueues[characteristic.uuid]?.removeFirstOrNull() ?: ByteArray(0)
+ }
override fun preferredWriteType(characteristic: BleCharacteristic): BleWriteType = BleWriteType.WITH_RESPONSE
override suspend fun write(characteristic: BleCharacteristic, data: ByteArray, writeType: BleWriteType) {
+ writeException?.let { ex -> throw ex }
availableCharacteristics += characteristic.uuid
writes += FakeBleWrite(characteristic = characteristic, data = data.copyOf(), writeType = writeType)
}
diff --git a/core/testing/src/commonMain/kotlin/org/meshtastic/core/testing/FakeRadioTransport.kt b/core/testing/src/commonMain/kotlin/org/meshtastic/core/testing/FakeRadioTransport.kt
index 4928024260..47ed6b148d 100644
--- a/core/testing/src/commonMain/kotlin/org/meshtastic/core/testing/FakeRadioTransport.kt
+++ b/core/testing/src/commonMain/kotlin/org/meshtastic/core/testing/FakeRadioTransport.kt
@@ -22,6 +22,9 @@ import org.meshtastic.core.repository.RadioTransport
class FakeRadioTransport : RadioTransport {
val sentData = mutableListOf<ByteArray>()
var closeCalled = false
+ var closeCount = 0
+ private set
+
var keepAliveCalled = false
override fun handleSendToRadio(p: ByteArray) {
@@ -34,5 +37,6 @@ class FakeRadioTransport : RadioTransport {
override suspend fun close() {
closeCalled = true
+ closeCount++
}
}
diff --git a/core/testing/src/commonTest/kotlin/org/meshtastic/core/testing/FakeBleServiceFailureInjectionTest.kt b/core/testing/src/commonTest/kotlin/org/meshtastic/core/testing/FakeBleServiceFailureInjectionTest.kt
new file mode 100644
index 0000000000..aa5f1e002f
--- /dev/null
+++ b/core/testing/src/commonTest/kotlin/org/meshtastic/core/testing/FakeBleServiceFailureInjectionTest.kt
@@ -0,0 +1,131 @@
+/*
+ * Copyright (c) 2026 Meshtastic LLC
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program. If not, see <https://www.gnu.org/licenses/>.
+ */
+package org.meshtastic.core.testing
+
+import kotlinx.coroutines.flow.collect
+import kotlinx.coroutines.flow.first
+import kotlinx.coroutines.launch
+import kotlinx.coroutines.test.runTest
+import kotlinx.coroutines.withTimeoutOrNull
+import org.meshtastic.core.ble.BleCharacteristic
+import org.meshtastic.core.ble.BleWriteType
+import kotlin.test.Test
+import kotlin.test.assertFailsWith
+import kotlin.test.assertFalse
+import kotlin.test.assertNotNull
+import kotlin.test.assertTrue
+import kotlin.uuid.Uuid
+
+class FakeBleServiceFailureInjectionTest {
+ private val char = BleCharacteristic(Uuid.random())
+ private val service = FakeBleService()
+
+ @Test
+ fun writeExceptionIsPersistent() = runTest {
+ service.writeException = Exception("write-fail")
+ val ex1 = assertFailsWith<Exception> { service.write(char, ByteArray(0), BleWriteType.WITH_RESPONSE) }
+ assertTrue(ex1.message == "write-fail", "Expected write-fail message")
+ // Prove persistence β second write also throws because exception persists
+ val ex2 = assertFailsWith<Exception> { service.write(char, ByteArray(1), BleWriteType.WITH_RESPONSE) }
+ assertTrue(ex2.message == "write-fail", "Expected write-fail message on second call")
+ // tests must explicitly clear it
+ service.writeException = null
+ service.write(char, ByteArray(2), BleWriteType.WITH_RESPONSE)
+ assertTrue(service.writes.size == 1)
+ }
+
+ @Test
+ fun readExceptionIsThrownAndReset() = runTest {
+ service.readException = Exception("read-fail")
+ assertFailsWith<Exception> { service.read(char) }
+ assertTrue(service.read(char).isEmpty(), "Read should return empty after exception reset")
+ }
+
+ @Test
+ fun observeExceptionCausesFlowException() = runTest {
+ service.observeException = Exception("observe-fail")
+ val flow = service.observe(char)
+ assertFailsWith<Exception> { flow.collect() }
+ }
+
+ @Test
+ fun observeExceptionByCharacteristicOnlyFailsTargetCharacteristic() = runTest {
+ val otherChar = BleCharacteristic(Uuid.random())
+ service.observeExceptionsByCharacteristic[char.uuid] = Exception("target-observe-fail")
+
+ val ex = assertFailsWith<Exception> { service.observe(char).collect() }
+ assertTrue(ex.message == "target-observe-fail", "Expected target observe failure")
+
+ var subscribed = false
+ service.emitNotification(otherChar.uuid, byteArrayOf(1))
+ withTimeoutOrNull(100) { service.observe(otherChar) { subscribed = true }.first() }
+ assertTrue(subscribed, "Other characteristic should still subscribe normally")
+ }
+
+ @Test
+ fun observeBeforeSubscriptionExceptionDoesNotInvokeOnSubscription() = runTest {
+ var subscribed = false
+ service.observeBeforeSubscriptionExceptionByCharacteristic[char.uuid] = Exception("before-subscribe-fail")
+
+ val ex = assertFailsWith<Exception> { service.observe(char) { subscribed = true }.collect() }
+
+ assertTrue(ex.message == "before-subscribe-fail", "Expected before-subscription failure")
+ assertFalse(subscribed, "onSubscription must not run before the injected failure")
+ }
+
+ @Test
+ fun observeExceptionByCharacteristicInTwoArgAlsoThrowsBeforeSubscription() = runTest {
+ // observeExceptionsByCharacteristic is the 1-arg seam, but the 2-arg override must also treat it as a
+ // pre-readiness failure β otherwise the default onStart wrap would invoke onSubscription before the throw.
+ var subscribed = false
+ service.observeExceptionsByCharacteristic[char.uuid] = Exception("shared-observe-fail")
+
+ val ex = assertFailsWith<Exception> { service.observe(char) { subscribed = true }.collect() }
+
+ assertTrue(ex.message == "shared-observe-fail", "Expected the shared characteristic exception")
+ assertFalse(subscribed, "onSubscription must not run before a shared characteristic failure")
+ }
+
+ @Test
+ fun observeNeverSubscribeDoesNotInvokeOnSubscription() = runTest {
+ var subscribed = false
+ service.observeNeverSubscribeCharacteristics += char.uuid
+
+ withTimeoutOrNull(100) { service.observe(char) { subscribed = true }.collect() }
+
+ assertFalse(subscribed, "onSubscription must not run for never-subscribe characteristic")
+ }
+
+ @Test
+ fun observeNeverSubscribeStillExposesNotifications() = runTest {
+ // Even though onSubscription is suppressed, the returned flow is the bare SharedFlow, so emitNotification
+ // must still reach active collectors.
+ service.observeNeverSubscribeCharacteristics += char.uuid
+ var subscribed = false
+ var received: ByteArray? = null
+
+ val collector = launch { service.observe(char) { subscribed = true }.collect { received = it } }
+ testScheduler.advanceUntilIdle()
+ service.emitNotification(char.uuid, byteArrayOf(1, 2, 3))
+ testScheduler.advanceUntilIdle()
+ collector.cancel()
+
+ assertFalse(subscribed, "onSubscription must not run for never-subscribe characteristic")
+ assertNotNull(received, "Notifications must still flow through the bare SharedFlow")
+ assertTrue(received!!.contentEquals(byteArrayOf(1, 2, 3)), "Notification payload must be exposed verbatim")
+ }
+}
Served by rngit 1.5.0 - Generated in 0.18s